from typing import Any, NamedTuple, Optional, TypedDict, Unpack, reveal_type from aiohttp import web from data.database import Database from datamodels import DataModel from .lib import datamodelsv, ModelField routes = web.RouteTableDef() class StampPayload(TypedDict): stamp_id: int document_id: int stamp_type: str pos_x: int pos_y: int rotation: float class StampCreateParamsReq(TypedDict, total=True): document_id: int stamp_type: str pos_x: int pos_y: int rotation: float class StampCreateParams(StampCreateParamsReq, total=False): pass class StampEditParams(TypedDict, total=False): document_id: int stamp_type: str pos_x: int pos_y: int rotation: float fields = [ ModelField('stamp_id', int, False, False, False), ModelField('document_id', int, True, True, True), ModelField('stamp_type', int, True, True, True), ModelField('pos_x', int, True, True, True), ModelField('pos_y', int, True, True, True), ModelField('rotation', float, True, True, True), ] req_fields = {field.name for field in fields if field.required} edit_fields = {field.name for field in fields if field.can_edit} create_fields = {field.name for field in fields if field.can_create} class Stamp: def __init__(self, app: web.Application, row: DataModel.DocumentStamp): self.app = app self.data = app[datamodelsv] self.row = row @classmethod async def fetch_from_id(cls, app: web.Application, stamp_id: int) -> Optional['Stamp']: stamp = await app[datamodelsv].DocumentStamp.fetch(stamp_id) if stamp is None: return None return cls(app, stamp) @classmethod async def query( cls, app: web.Application, stamp_id: Optional[int] = None, document_id: Optional[int] = None, stamp_type: Optional[str] = None, ): data = app[datamodelsv] query_args = {} if stamp_id is not None: query_args['stamp_id'] = stamp_id if document_id is not None: query_args['document_id'] = document_id if stamp_type is not None: typerows = await data.StampType.table.fetch_rows_where(stamp_type_name=stamp_type) typeids = [row.stamp_type_id for row in typerows] if not typeids: return [] query_args['stamp_type'] = typeids results = await data.DocumentStamp.table.fetch_rows_where(**query_args) return [cls(app, row) for row in results] @classmethod async def create( cls, app: web.Application, **kwargs: Unpack[StampCreateParams] ): data = app[datamodelsv] stamp_type = kwargs['stamp_type'] # Get the stamp_type rows = await data.StampType.table.fetch_rows_where(stamp_type_name=stamp_type) if not rows: # Create the stamp type row = await data.StampType.create(stamp_type_name=stamp_type) else: row = rows[0] stamprow = await data.DocumentStamp.create( document_id=kwargs['document_id'], stamp_type=row.stamp_type_id, position_x=int(kwargs['pos_x']), position_y=int(kwargs['pos_y']), rotation=float(kwargs['rotation']) ) return cls(app, stamprow) async def prepare(self) -> StampPayload: typerow = await self.data.StampType.fetch(self.row.stamp_type) assert typerow is not None results: StampPayload = { 'stamp_id': self.row.stamp_id, 'document_id': self.row.document_id, 'stamp_type': typerow.stamp_type_name, 'pos_x': self.row.position_x, 'pos_y': self.row.position_y, 'rotation': self.row.rotation, } return results async def edit( self, **kwargs: Unpack[StampEditParams] ): data = self.data row = self.row edit_args = {} if stamp_type := kwargs.get('stamp_type'): # Get the stamp_type rows = await data.StampType.table.fetch_rows_where(stamp_type_name=stamp_type) if not rows: # Create the stamp type row = await data.StampType.create(stamp_type_name=stamp_type) else: row = rows[0] edit_args['stamp_type'] = row.stamp_type_id simple_keys = { 'document_id': 'document_id', 'pos_x': 'position_x', 'pos_y': 'position_y', 'rotation': 'rotation' } for editkey, datakey in simple_keys.values(): if editkey in kwargs: edit_args[datakey] = kwargs[editkey] await self.row.update( **edit_args ) async def delete(self) -> StampPayload: payload = await self.prepare() await self.row.delete() return payload @routes.view('/stamps') class StampsView(web.View): async def get(self): request = self.request # Decode request parameters to filter args filter_params = {} keys = ['stamp_id', 'document_id', 'stamp_type'] for key in keys: if key in request.query: filter_params[key] = request.query[key] elif key in request: filter_params[key] = request[key] stamps = await Stamp.query(request.app, **filter_params) payload = [await stamp.prepare() for stamp in stamps] return web.json_response(payload) async def create_one(self, params: StampCreateParams): if extra := next((key for key in params if key not in create_fields), None): raise web.HTTPBadRequest(text=f"Invalid key '{extra}' passed to stamp creation.") if missing := next((key for key in req_fields if key not in params), None): raise web.HTTPBadRequest(text=f"Stamp params missing required key '{missing}'.") # This still doesn't guarantee that the values are of the correct type, but good enough. stamp = await Stamp.create(self.request.app, **params) return stamp async def post(self): request = self.request params = await request.json() for key in create_fields: if key in request: params.setdefault(key, request[key]) stamp = await self.create_one(params) payload = await stamp.prepare() return web.json_response(payload) async def put(self): request = self.request from_request = {key: request[key] for key in create_fields if key in request} argslist = await request.json() payloads = [] for args in argslist: stamp = await self.create_one(from_request | args) payload = await stamp.prepare() payloads.append(payload) return web.json_response(payloads) @routes.view('/stamps/{stamp_id}') class StampView(web.View): async def resolve_stamp(self): request = self.request stamp_id = request.match_info['stamp_id'] stamp = await Stamp.fetch_from_id(request.app, int(stamp_id)) if stamp is None: raise web.HTTPNotFound(text="No stamp exists with the given ID.") return stamp async def get(self): stamp = await self.resolve_stamp() payload = await stamp.prepare() return web.json_response(payload) async def patch(self): stamp = await self.resolve_stamp() params = await self.request.json() edit_data = {} for key, value in params.items(): if key not in edit_fields: raise web.HTTPBadRequest(text=f"You cannot update field '{key}' of Stamp!") edit_data[key] = value for key in edit_fields: if key in self.request: edit_data.setdefault(key, self.request[key]) await stamp.edit(**edit_data) payload = await stamp.prepare() return web.json_response(payload) async def delete(self): stamp = await self.resolve_stamp() payload = await stamp.delete() return web.json_response(payload)