TheLovinator/discord-rss-bot
1
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Add type hints

Browse Source
This commit is contained in:
Joakim Hellsén
2022-12-05 19:12:23 +01:00
parent 849cb17e95
commit 4c5faa2181
3 changed files with 46 additions and 40 deletions
Download Patch File Download Diff File Expand all files Collapse all files

43
discord_rss_bot/main.py
View File

@ -29,37 +29,41 @@ Functions:
import enum
import sys
from functools import cache
from typing import Iterable
import uvicorn
from apscheduler.schedulers.background import BackgroundScheduler
from fastapi import FastAPI, Form, HTTPException, Request
from fastapi.responses import FileResponse, HTMLResponse
from fastapi.staticfiles import StaticFiles
from fastapi.templating import Jinja2Templates
from reader import ResourceNotFoundError
from reader import Entry, EntryCounts, Feed, FeedCounts, ResourceNotFoundError
from tomlkit.toml_document import TOMLDocument
from discord_rss_bot.feeds import add_feed, send_to_discord, update_feed
from discord_rss_bot.feeds import IfFeedError, add_feed, send_to_discord, update_feed
from discord_rss_bot.settings import logger, read_settings_file, reader
from discord_rss_bot.webhooks import set_hook_by_name
app = FastAPI()
templates = Jinja2Templates(directory="templates")
app: FastAPI = FastAPI()
app.mount("/static", StaticFiles(directory="static"), name="static")
templates: Jinja2Templates = Jinja2Templates(directory="templates")
@app.post("/check", response_class=HTMLResponse)
def check_feed(request: Request, feed_url: str = Form()):
"""Check all feeds"""
reader.update_feeds()
entry = reader.get_entries(feed=feed_url, read=False)
entry: Iterable[Entry] = reader.get_entries(feed=feed_url, read=False)
send_to_discord(entry)
logger.info(f"Get feed: {feed_url}")
feed = reader.get_feed(feed_url)
feed: Feed = reader.get_feed(feed_url)
return templates.TemplateResponse("feed.html", {"request": request, "feed": feed})
@app.post("/add")
async def create_feed(feed_url: str = Form(), webhook_dropdown: str = Form()):
async def create_feed(feed_url: str = Form(), webhook_dropdown: str = Form()) -> HTTPException | dict[str, str]:
"""
Add a feed to the database.
@ -74,10 +78,10 @@ async def create_feed(feed_url: str = Form(), webhook_dropdown: str = Form()):
logger.info(f"Webhook: {webhook_dropdown}")
# Update a single feed. The feed will be updated even if updates are disabled for it.
updated_feed = update_feed(feed_url, webhook_dropdown)
updated_feed: IfFeedError = update_feed(feed_url, webhook_dropdown)
# Add a new feed to the database.
added_feed = add_feed(feed_url, webhook_dropdown)
added_feed: IfFeedError = add_feed(feed_url, webhook_dropdown)
if updated_feed.error or added_feed.error:
error_dict = {
@ -95,7 +99,7 @@ async def create_feed(feed_url: str = Form(), webhook_dropdown: str = Form()):
):
return set_hook_by_name(name=webhook_dropdown, feed_url=feed_url)
new_tag = reader.get_tag(feed_url, "webhook")
new_tag: str = str(reader.get_tag(feed_url, "webhook"))
logger.info(f"New tag: {new_tag}")
return {"feed_url": str(feed_url), "status": "added"}
@ -103,7 +107,7 @@ async def create_feed(feed_url: str = Form(), webhook_dropdown: str = Form()):
def create_list_of_webhooks():
"""List with webhooks."""
logger.info("Creating list with webhooks.")
settings = read_settings_file()
settings: TOMLDocument = read_settings_file()
list_of_webhooks = dict()
for hook in settings["webhooks"]:
logger.info(f"Webhook name: {hook} with URL: {settings['webhooks'][hook]}")
@ -115,7 +119,7 @@ def create_list_of_webhooks():
@cache
@app.get("/favicon.ico", include_in_schema=False)
async def favicon():
async def favicon() -> FileResponse:
"""Return favicon."""
return FileResponse("static/favicon.ico")
@ -184,12 +188,12 @@ def make_context_index(request) -> dict:
logger.info(f"Webhook name: {hook.name}")
feed_list = list()
feeds = reader.get_feeds()
feeds: Iterable[Feed] = reader.get_feeds()
for feed in feeds:
feed_list.append(feed)
feed_count = reader.get_feed_counts()
entry_count = reader.get_entry_counts()
feed_count: FeedCounts = reader.get_feed_counts()
entry_count: EntryCounts = reader.get_entry_counts()
context = {
"request": request,
"feeds": feed_list,
@ -225,17 +229,18 @@ def startup():
"""This is called when the server starts.
It reads the settings file and starts the scheduler."""
settings = read_settings_file()
settings: TOMLDocument = read_settings_file()
if not settings["webhooks"]:
logger.critical("No webhooks found in settings file.")
sys.exit()
for key in settings["webhooks"]:
webhooks = settings["webhooks"]
for key in webhooks:
logger.info(f"Webhook name: {key} with URL: {settings['webhooks'][key]}")
scheduler = BackgroundScheduler()
scheduler: BackgroundScheduler = BackgroundScheduler()
scheduler.start()
if __name__ == "__main__":
uvicorn.run("main:app", log_level="debug")
uvicorn.run("main:app", log_level="debug", reload=True)

34
discord_rss_bot/settings.py
View File

@ -21,14 +21,16 @@ import os
from pathlib import Path
from platformdirs import user_data_dir
from reader import make_reader
from tomlkit import TOMLDocument, comment, document, parse, table
from reader import Reader, make_reader
from tomlkit import comment, document, parse, table
from tomlkit.items import Table
from tomlkit.toml_document import TOMLDocument
logging.basicConfig(level=logging.DEBUG, format="[%(asctime)s] [%(funcName)s:%(lineno)d] %(message)s")
logger = logging.getLogger(__name__)
logger: logging.Logger = logging.getLogger(__name__)
# For get_data_dir()
data_directory = user_data_dir(appname="discord_rss_bot", appauthor="TheLovinator", roaming=True)
data_directory: str = user_data_dir(appname="discord_rss_bot", appauthor="TheLovinator", roaming=True)
def create_settings_file(settings_file) -> None:
@ -37,18 +39,16 @@ def create_settings_file(settings_file) -> None:
# [webhooks]
# Both options are commented out by default.
webhooks = table()
webhooks: Table = table()
webhooks.add(comment('"First webhook" = "https://discord.com/api/webhooks/1234567890/abcdefghijklmnopqrstuvwxyz"'))
webhooks.add(
comment('"Second webhook" = "https://discord.com/api/webhooks/1234567890/abcdefghijklmnopqrstuvwxyz"')
)
webhooks.add(comment('"Second webhook" = "https://discord.com/api/webhooks/1234567890/abcdefghijklmnopqrstuvwxyz"'))
# [database]
# Option is commented out by default.
database = table()
database: Table = table()
database.add(comment('"location" = "/path/to/database/file"'))
doc = document()
doc: TOMLDocument = document()
doc.add("webhooks", webhooks)
doc.add("database", database)
@ -74,7 +74,7 @@ def get_data_dir(data_dir: str = data_directory) -> Path:
logger.info(f"Using custom data directory: {data_dir}")
# Use the environment variable if it exists instead of the default app dir.
data_dir = os.getenv("DATA_DIR") or data_dir
data_dir: str = os.getenv("DATA_DIR") or data_dir
logger.debug(f"Data directory: {data_dir}")
@ -97,11 +97,11 @@ def get_db_file(custom_db_name: str = "db.sqlite") -> Path:
logger.info(f"Using custom database file: {custom_db_name}")
# Store the database file in the data directory
data_dir = get_data_dir()
data_dir: Path = get_data_dir()
db_location: Path = Path(os.path.join(data_dir, custom_db_name))
# Use the environment variable if it exists instead of the default db name.
db_file = os.getenv("DATABASE_LOCATION") or db_location
db_file: str | Path = os.getenv("DATABASE_LOCATION") or db_location
logger.debug(f"Database file: {db_file}")
return Path(db_file)
@ -120,11 +120,11 @@ def read_settings_file(custom_settings_name: str = "settings.toml") -> TOMLDocum
logger.info(f"Using custom name for settings file: {custom_settings_name}")
# Store the database file in the data directory
data_dir = get_data_dir()
data_dir: Path = get_data_dir()
settings_file_location: Path = Path(os.path.join(data_dir, custom_settings_name))
# Use the environment variable if it exists instead of the default db name.
settings_file = os.getenv("SETTINGS_FILE_LOCATION") or settings_file_location
settings_file: str | Path = os.getenv("SETTINGS_FILE_LOCATION") or settings_file_location
logger.debug(f"Settings file: {settings_file}")
# Create the settings file if it doesn't exist
@ -132,10 +132,10 @@ def read_settings_file(custom_settings_name: str = "settings.toml") -> TOMLDocum
create_settings_file(settings_file)
with open(settings_file, encoding="utf-8") as f:
data = parse(f.read())
data: TOMLDocument = parse(f.read())
logger.debug(f"Contents of settings file: {data}")
return data
reader = make_reader(str(get_db_file()))
reader: Reader = make_reader(str(get_db_file()))

9
discord_rss_bot/webhooks.py
View File

@ -1,10 +1,11 @@
from fastapi import HTTPException
from reader import ResourceNotFoundError
from tomlkit.toml_document import TOMLDocument
from discord_rss_bot.settings import logger, read_settings_file, reader
def set_hook_by_name(name: str, feed_url: str) -> None or HTTPException:
def set_hook_by_name(name: str, feed_url: str) -> HTTPException:
"""Set a webhook by name.
Args:
@ -14,13 +15,13 @@ def set_hook_by_name(name: str, feed_url: str) -> None or HTTPException:
Returns:
HTTPException: The HTTP exception if the webhook was not found, otherwise None.
"""
settings = read_settings_file()
settings: TOMLDocument = read_settings_file()
logger.debug(f"Webhook name: {name} with URL: {settings['webhooks'][name]}")
webhook_url = settings["webhooks"][name]
webhook_url: str = str(settings["webhooks"][name])
try:
reader.set_tag(feed_url, "webhook", webhook_url)
except ResourceNotFoundError as e:
error_msg = f"ResourceNotFoundError: Could not set webhook: {e}"
error_msg: str = f"ResourceNotFoundError: Could not set webhook: {e}"
logger.error(error_msg, exc_info=True)
return HTTPException(status_code=500, detail=error_msg)