43 lines
1.4 KiB
Python
43 lines
1.4 KiB
Python
from typing import Optional
|
|
from .data import Koan, KoanInfo, KoansData
|
|
from .lib import utc_now
|
|
|
|
|
|
class KoanRegistry:
|
|
def __init__(self, data: KoansData):
|
|
self.data = data
|
|
|
|
async def init(self):
|
|
await self.data.init()
|
|
|
|
async def get_community_koans(self, communityid: int) -> list[KoanInfo]:
|
|
return await KoanInfo.fetch_where(communityid=communityid, is_deleted=False)
|
|
|
|
async def get_koaninfo(self, koanid: int) -> Optional[KoanInfo]:
|
|
return await KoanInfo.fetch(koanid)
|
|
|
|
async def get_koan(self, koanid: int) -> Optional[Koan]:
|
|
return await Koan.fetch(koanid)
|
|
|
|
async def get_koan_label(self, communityid: int, label: int) -> Optional[KoanInfo]:
|
|
results = await KoanInfo.fetch_where(communityid=communityid, koanlabel=label, is_deleted=False)
|
|
return results[0] if results else None
|
|
|
|
async def create_koan(
|
|
self,
|
|
communityid: int,
|
|
content: str,
|
|
created_by: Optional[int] = None
|
|
) -> KoanInfo:
|
|
koan = await Koan.create(
|
|
content=content,
|
|
communityid=communityid,
|
|
created_by=created_by,
|
|
)
|
|
info = await KoanInfo.fetch(koan.koanid)
|
|
assert info is not None
|
|
return info
|
|
|
|
async def delete_koan(self, koanid: int):
|
|
await self.data.koans.update_where(koanid=koanid).set(deleted_at=utc_now())
|