diff --git a/discord_rss_bot/feeds.py b/discord_rss_bot/feeds.py index b7079bd..2499016 100644 --- a/discord_rss_bot/feeds.py +++ b/discord_rss_bot/feeds.py @@ -21,9 +21,12 @@ Exceptions: Used in send_to_discord(). If no webhook found, it will raise NoWebhookFoundError. """ +from typing import Any, Iterable + from discord_webhook import DiscordWebhook from pydantic import BaseModel from reader import ( + Entry, EntryNotFoundError, FeedExistsError, FeedNotFoundError, @@ -60,10 +63,10 @@ class NoWebhookFoundError(Exception): Used in send_to_discord().""" - def __init__(self, message): + def __init__(self, message) -> None: self.message = message - def __str__(self): + def __str__(self) -> Any: return self.message @@ -122,7 +125,7 @@ def check_feed(feed_url: str) -> None: send_to_discord(entry) -def check_feeds() -> None: +def send_to_discord(feed=None) -> None: """Update all feeds and send all the entries that are unread to Discord. We don't need to mark entries as read here, because send_to_discord() does that when sending entries to Discord @@ -132,7 +135,6 @@ def check_feeds() -> None: send_to_discord() -def send_to_discord(feed=None): """ Send entries to Discord. @@ -148,9 +150,9 @@ def send_to_discord(feed=None): Response: The response from the webhook. """ if feed is None: - entries = reader.get_entries(read=False) + entries: Iterable[Entry] = reader.get_entries(read=False) else: - entries = reader.get_entries(feed=feed, read=False) + entries: Iterable[Entry] = reader.get_entries(feed=feed, read=False) if not entries: logger.info("No entries to send") @@ -169,7 +171,7 @@ def send_to_discord(feed=None): raise try: - webhook_url = str(reader.get_tag(entry.feed.url, "webhook")) + webhook_url: str = str(reader.get_tag(entry.feed.url, "webhook")) except TagNotFoundError: logger.error("Tag not found", exc_info=True) raise @@ -182,9 +184,9 @@ def send_to_discord(feed=None): raise NoWebhookFoundError(f"No webhook found for feed: {entry.feed.url}") logger.debug(f"Sending to webhook: {webhook_url}") - webhook_message = f":robot: :mega: {entry.title}\n{entry.link}" - webhook = DiscordWebhook(url=webhook_url, content=webhook_message, rate_limit_retry=True) - response = webhook.execute() + webhook_message: str = f":robot: :mega: {entry.title}\n{entry.link}" + webhook: DiscordWebhook = DiscordWebhook(url=webhook_url, content=webhook_message, rate_limit_retry=True) + response: Response = webhook.execute() if not response.ok: logger.error(f"Error: {response.status_code} {response.reason}") reader.set_entry_read(entry, False) diff --git a/discord_rss_bot/main.py b/discord_rss_bot/main.py index e942262..4dbe133 100644 --- a/discord_rss_bot/main.py +++ b/discord_rss_bot/main.py @@ -29,7 +29,7 @@ Functions: import enum import sys from functools import cache -from typing import Iterable +from typing import Any, Iterable import uvicorn from apscheduler.schedulers.background import BackgroundScheduler @@ -38,6 +38,7 @@ from fastapi.responses import FileResponse, HTMLResponse from fastapi.staticfiles import StaticFiles from fastapi.templating import Jinja2Templates from reader import EntryCounts, Feed, FeedCounts, ResourceNotFoundError +from starlette.templating import _TemplateResponse from tomlkit.toml_document import TOMLDocument from discord_rss_bot.feeds import IfFeedError, add_feed, send_to_discord, update_feed @@ -50,7 +51,7 @@ templates: Jinja2Templates = Jinja2Templates(directory="templates") @app.post("/check", response_class=HTMLResponse) -def check_feed(request: Request, feed_url: str = Form()): +def check_feed(request: Request, feed_url: str = Form()) -> _TemplateResponse: """Check all feeds""" reader.update_feeds() send_to_discord(feed_url) @@ -83,7 +84,7 @@ async def create_feed(feed_url: str = Form(), webhook_dropdown: str = Form()) -> updated_feed: IfFeedError = update_feed(feed_url, webhook_dropdown) if updated_feed.error or added_feed.error: - error_dict = { + error_dict: dict[str, Any] = { "error": updated_feed.error, "feed": updated_feed.feed_url, "webhook": updated_feed.webhook, @@ -103,7 +104,7 @@ async def create_feed(feed_url: str = Form(), webhook_dropdown: str = Form()) -> return {"feed_url": str(feed_url), "status": "added"} -def create_list_of_webhooks(): +def create_list_of_webhooks() -> enum.EnumMeta: """List with webhooks.""" logger.info("Creating list with webhooks.") settings: TOMLDocument = read_settings_file() @@ -124,7 +125,7 @@ async def favicon() -> FileResponse: @app.get("/add", response_class=HTMLResponse) -def get_add(request: Request): +def get_add(request: Request) -> _TemplateResponse: """ Page for adding a new feed. @@ -139,7 +140,7 @@ def get_add(request: Request): @app.get("/feed/{feed_url:path}", response_class=HTMLResponse) -async def get_feed(feed_url: str, request: Request): +async def get_feed(feed_url: str, request: Request) -> _TemplateResponse: """ Get a feed by URL. @@ -153,12 +154,12 @@ async def get_feed(feed_url: str, request: Request): # Convert the URL to a valid URL. logger.info(f"Got 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.get("/", response_class=HTMLResponse) -def index(request: Request): +def index(request: Request) -> _TemplateResponse: """ This is the root of the website. @@ -184,18 +185,18 @@ def make_context_index(request) -> dict: dict: The context. """ - hooks = create_list_of_webhooks() + hooks: enum.EnumMeta = create_list_of_webhooks() for hook in hooks: logger.info(f"Webhook name: {hook.name}") - feed_list = list() + feed_list: list[Feed] = list() feeds: Iterable[Feed] = reader.get_feeds() for feed in feeds: feed_list.append(feed) feed_count: FeedCounts = reader.get_feed_counts() entry_count: EntryCounts = reader.get_entry_counts() - context = { + context: dict[str, Any] = { "request": request, "feeds": feed_list, "feed_count": feed_count, @@ -206,7 +207,7 @@ def make_context_index(request) -> dict: @app.post("/remove", response_class=HTMLResponse) -async def remove_feed(request: Request, feed_url: str = Form()): +async def remove_feed(request: Request, feed_url: str = Form()) -> _TemplateResponse: """ Get a feed by URL. @@ -219,14 +220,14 @@ async def remove_feed(request: Request, feed_url: str = Form()): """ logger.info(f"Get feed: {feed_url}") - feed = reader.get_feed(feed_url) + feed: Feed = reader.get_feed(feed_url) reader.delete_feed(feed_url) return templates.TemplateResponse("index.html", {"request": request, "feed": feed}) @app.on_event("startup") -def startup(): +def startup() -> None: """This is called when the server starts. It reads the settings file and starts the scheduler."""