Add a function for get_webhooks()
This commit is contained in:
@ -51,7 +51,7 @@ from starlette.templating import _TemplateResponse
|
|||||||
|
|
||||||
from discord_rss_bot.feeds import send_to_discord
|
from discord_rss_bot.feeds import send_to_discord
|
||||||
from discord_rss_bot.search import create_html_for_search_results
|
from discord_rss_bot.search import create_html_for_search_results
|
||||||
from discord_rss_bot.settings import get_reader
|
from discord_rss_bot.settings import get_reader, get_webhooks
|
||||||
|
|
||||||
app: FastAPI = FastAPI()
|
app: FastAPI = FastAPI()
|
||||||
app.mount("/static", StaticFiles(directory="discord_rss_bot/static"), name="static")
|
app.mount("/static", StaticFiles(directory="discord_rss_bot/static"), name="static")
|
||||||
@ -94,12 +94,7 @@ async def add_webhook(webhook_name: str = Form(), webhook_url: str = Form()) ->
|
|||||||
clean_webhook_url: str = webhook_url.strip()
|
clean_webhook_url: str = webhook_url.strip()
|
||||||
|
|
||||||
# Get current webhooks from the database if they exist otherwise use an empty list.
|
# Get current webhooks from the database if they exist otherwise use an empty list.
|
||||||
webhooks: list[dict[str, str]] = []
|
webhooks = get_webhooks(reader)
|
||||||
if reader.get_tags(()) is not None:
|
|
||||||
for tag in reader.get_tag_keys(()):
|
|
||||||
if tag == "webhooks":
|
|
||||||
webhooks = reader.get_tag((), "webhooks")
|
|
||||||
break
|
|
||||||
|
|
||||||
# Only add the webhook if it doesn't already exist.
|
# Only add the webhook if it doesn't already exist.
|
||||||
if not any(webhook["name"] == clean_webhook_name for webhook in webhooks):
|
if not any(webhook["name"] == clean_webhook_name for webhook in webhooks):
|
||||||
@ -134,12 +129,7 @@ async def delete_webhook(webhook_url: str = Form()) -> RedirectResponse | dict[s
|
|||||||
clean_webhook_url: str = webhook_url.strip()
|
clean_webhook_url: str = webhook_url.strip()
|
||||||
|
|
||||||
# Get current webhooks from the database if they exist otherwise use an empty list.
|
# Get current webhooks from the database if they exist otherwise use an empty list.
|
||||||
webhooks: list[dict[str, str]] = []
|
webhooks = get_webhooks(reader)
|
||||||
if reader.get_tags(()) is not None:
|
|
||||||
for tag in reader.get_tag_keys(()):
|
|
||||||
if tag == "webhooks":
|
|
||||||
webhooks = reader.get_tag((), "webhooks")
|
|
||||||
break
|
|
||||||
|
|
||||||
# Only add the webhook if it doesn't already exist.
|
# Only add the webhook if it doesn't already exist.
|
||||||
for webhook in webhooks:
|
for webhook in webhooks:
|
||||||
|
@ -45,3 +45,22 @@ def get_reader(custom_location: str = "") -> Reader:
|
|||||||
"""
|
"""
|
||||||
db_location: str = get_db_location(custom_location)
|
db_location: str = get_db_location(custom_location)
|
||||||
return make_reader(url=db_location)
|
return make_reader(url=db_location)
|
||||||
|
|
||||||
|
|
||||||
|
def get_webhooks(reader: Reader) -> list[dict[str, str]]:
|
||||||
|
"""
|
||||||
|
Get current webhooks from the database if they exist otherwise use an empty list.
|
||||||
|
|
||||||
|
Args:
|
||||||
|
reader: The reader to use.
|
||||||
|
|
||||||
|
Returns:
|
||||||
|
list[dict[str, str]]: The webhooks.
|
||||||
|
"""
|
||||||
|
webhooks: list[dict[str, str]] = []
|
||||||
|
if reader.get_tags(()) is not None:
|
||||||
|
for tag in reader.get_tag_keys(()):
|
||||||
|
if tag == "webhooks":
|
||||||
|
webhooks = reader.get_tag((), "webhooks")
|
||||||
|
break
|
||||||
|
return webhooks
|
||||||
|
Reference in New Issue
Block a user