You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.

53 lines
1.4 KiB

# stdlib
from typing import Union
# internal files
import db_models
from db_models import db
class DBAssetStoreManager():
def __init__(self):
async def get_asset_url(self, id) -> Union[None, str]:
Get an asset's url by the ID.
Returns None if the ID is not found.
asset = await db_models.AssetStore.get(id)
if asset: # if this isn't none (scalar returns none if no results)
return asset.url
return None
async def store_asset(self, id, url) -> str:
Store or update an asset by ID.
Returns the URL of the asset.
await db_models.AssetStore.create(id=id, url=url)
return url
class DBModerationManager():
def __init__(self):
self._mod_cache = {} # Internal mod cache. This is used to prevent unneccesary database calls.
def get_rank_from_cache(self, guild_id, user_id):
def get_moderator_status(self, guild_id, user_id) -> db_models.ModRankEnum:
def add_moderator(self, guild_id, user_id, rank: db_models.ModRankEnum):
class DBOperationsManager():
def __init__(self):
self.asset_store = DBAssetStoreManager()
async def connect_database(self, db_path):
await db.set_bind(db_path)
async def disconnect_database(self):
await db.pop_bind().close()