61 lines
1.8 KiB
Python
61 lines
1.8 KiB
Python
from collections.abc import Mapping
|
|
from datetime import date
|
|
from typing import Optional
|
|
|
|
from app.dao.child import ChildDAO
|
|
from app.service.im import ensure_parent_child_conversation
|
|
|
|
|
|
class ChildService:
|
|
def __init__(self, db):
|
|
self.db = db
|
|
self.dao = ChildDAO(db)
|
|
|
|
def create(
|
|
self,
|
|
user_id: int,
|
|
child_name: str,
|
|
child_gender: int = 2,
|
|
child_birthday: Optional[date] = None,
|
|
) -> Mapping:
|
|
try:
|
|
child_id = self.dao.create(
|
|
user_id,
|
|
child_name,
|
|
child_gender,
|
|
child_birthday,
|
|
auto_commit=False,
|
|
)
|
|
ensure_parent_child_conversation(
|
|
self.db,
|
|
parent_user_id=user_id,
|
|
child_id=child_id,
|
|
)
|
|
self.db.commit()
|
|
return self.dao.get_by_id(child_id)
|
|
except Exception:
|
|
self.db.rollback()
|
|
raise
|
|
|
|
def list_children(self, user_id: int, limit: int = 20, cursor: int = None) -> tuple[list, bool]:
|
|
rows = self.dao.list_by_parent(user_id, limit, cursor)
|
|
has_more = len(rows) > limit
|
|
rows = rows[:limit]
|
|
return rows, has_more
|
|
|
|
def get(self, child_id: int) -> Optional[Mapping]:
|
|
return self.dao.get_by_id(child_id)
|
|
|
|
def update(
|
|
self,
|
|
child_id: int,
|
|
user_id: int,
|
|
child_name: Optional[str] = None,
|
|
child_gender: Optional[int] = None,
|
|
child_birthday: Optional[date] = None,
|
|
) -> Mapping:
|
|
if not self.dao.has_access(child_id, user_id):
|
|
raise PermissionError("No access to this child")
|
|
self.dao.update(child_id, child_name, child_gender, child_birthday)
|
|
return self.dao.get_by_id(child_id)
|