mirror of
https://github.com/revanced/revanced-releases-api.git
synced 2025-04-30 06:24:27 +02:00
0ce5780a4e
* refactor: import routers from old branch * refactor: import InternalCache removal * refactor: move routes into dedicated routers * fix: fixes entrypoint * refactor: add documentation and bump libs
102 lines
3.5 KiB
Python
102 lines
3.5 KiB
Python
import toml
|
|
from redis import asyncio as aioredis
|
|
|
|
import app.utils.Logger as Logger
|
|
from app.utils.Generators import Generators
|
|
from app.models.AnnouncementModels import AnnouncementCreateModel
|
|
from app.utils.RedisConnector import RedisConnector
|
|
|
|
config: dict = toml.load("config.toml")
|
|
|
|
class Announcements:
|
|
"""Implements the announcements class for the ReVanced API"""
|
|
|
|
redis = RedisConnector.connect(config['announcements']['database'])
|
|
|
|
AnnouncementsLogger = Logger.AnnouncementsLogger()
|
|
|
|
generators = Generators()
|
|
|
|
async def store(self, announcement: AnnouncementCreateModel, author: str) -> bool:
|
|
"""Store an announcement in the database
|
|
|
|
Args:
|
|
announcement (AnnouncementCreateModel): Pydantic model of the announcement
|
|
|
|
Returns:
|
|
str | bool: UUID of the announcement or False if the announcement wasn't stored successfully
|
|
"""
|
|
|
|
announcement_id: str = "announcement"
|
|
|
|
timestamp = await self.generators.generate_timestamp()
|
|
|
|
announcement_payload: dict[str, str | int] = {}
|
|
|
|
announcement_payload['created_at'] = timestamp
|
|
announcement_payload['author'] = author
|
|
announcement_payload['type'] = announcement.type
|
|
announcement_payload['title'] = announcement.title
|
|
announcement_payload['content'] = announcement.content
|
|
|
|
try:
|
|
await self.redis.json().set(announcement_id, '$', announcement_payload)
|
|
await self.AnnouncementsLogger.log("SET", None, announcement_id)
|
|
except aioredis.RedisError as e:
|
|
await self.AnnouncementsLogger.log("SET", e)
|
|
raise e
|
|
|
|
return True
|
|
|
|
async def exists(self) -> bool:
|
|
"""Check if an announcement exists in the database
|
|
|
|
Returns:
|
|
bool: True if the announcement exists, False otherwise
|
|
"""
|
|
try:
|
|
if await self.redis.exists("announcement"):
|
|
await self.AnnouncementsLogger.log("EXISTS", None, "announcement")
|
|
return True
|
|
else:
|
|
await self.AnnouncementsLogger.log("EXISTS", None, "announcement")
|
|
return False
|
|
except aioredis.RedisError as e:
|
|
await self.AnnouncementsLogger.log("EXISTS", e)
|
|
raise e
|
|
|
|
async def get(self) -> dict:
|
|
"""Get a announcement from the database
|
|
|
|
Returns:
|
|
dict: Dict of the announcement or an empty dict if the announcement doesn't exist
|
|
"""
|
|
|
|
if await self.exists():
|
|
try:
|
|
announcement: dict[str, str | int] = await self.redis.json().get("announcement")
|
|
await self.AnnouncementsLogger.log("GET", None, "announcement")
|
|
except aioredis.RedisError as e:
|
|
await self.AnnouncementsLogger.log("GET", e)
|
|
return {}
|
|
return announcement
|
|
else:
|
|
return {}
|
|
|
|
async def delete(self) -> bool:
|
|
"""Delete an announcement from the database
|
|
|
|
Returns:
|
|
bool: True if the announcement was deleted successfully, False otherwise
|
|
"""
|
|
|
|
if await self.exists():
|
|
try:
|
|
await self.redis.delete("announcement")
|
|
await self.AnnouncementsLogger.log("DELETE", None, "announcement")
|
|
except aioredis.RedisError as e:
|
|
await self.AnnouncementsLogger.log("DELETE", e)
|
|
return False
|
|
return True
|
|
else:
|
|
return False |