import logging from datetime import datetime from typing import Any, Literal, NamedTuple, Optional, Self, TypedDict, Unpack, overload, reveal_type, List from aiohttp import web from data import Condition, condition from data.conditions import NULL from data.queries import JOINTYPE from datamodels import DataModel, EventType from modules.profiles.data import ProfileData from .lib import ModelField, datamodelsv, dbvar, profiledatav from .specimens import Specimen, SpecimenPayload routes = web.RouteTableDef() logger = logging.getLogger(__name__) class Event: def __init__(self, app: web.Application, row: DataModel.EventDetails): self.app = app self.data = app[datamodelsv] self.row = row @classmethod async def fetch_from_id(cls, app: web.Application, event_id: int): data = app[datamodelsv] row = await data.EventDetails.fetch(int(event_id)) return cls(app, row) if row is not None else None @classmethod async def query( cls, app: web.Application, event_id: Optional[str] = None, document_id: Optional[str] = None, document_seal: Optional[str] = None, user_id: Optional[str] = None, user_name: Optional[str] = None, occurred_before: Optional[str] = None, occurred_after: Optional[str] = None, created_before: Optional[str] = None, created_after: Optional[str] = None, event_type: Optional[str] = None, ) -> List[Self]: data = app[datamodelsv] EventD = data.EventDetails conds = [] if event_id is not None: conds.append(EventD.event_id == int(event_id)) if document_id is not None: conds.append(EventD.document_id == int(document_id)) if document_seal is not None: conds.append(EventD.document_seal == int(document_seal)) if user_id is not None: conds.append(EventD.user_id == int(user_id)) if user_name is not None: conds.append(EventD.user_name == user_name) if created_before is not None: cbefore = datetime.fromisoformat(created_before) conds.append(EventD.created_at <= cbefore) if created_after is not None: cafter = datetime.fromisoformat(created_after) conds.append(EventD.created_at >= cafter) if occurred_before is not None: before = datetime.fromisoformat(occurred_before) conds.append(EventD.occurred_at <= before) if occurred_after is not None: after = datetime.fromisoformat(occurred_after) conds.append(EventD.occurred_at >= after) if event_type is not None: ekey = (event_type.lower().strip(),) if ekey not in EventType: raise web.HTTPBadRequest(text=f"Unknown event type '{event_type}'") conds.append(EventD.event_type == EventType(ekey)) rows = await EventD.fetch_where(*conds).order_by(EventD.occurred_at) return [cls(app, row) for row in rows] @classmethod async def validate_create_params(cls, params): if 'event_type' not in params: raise web.HTTPBadRequest(text="Event creation missing required field 'event_type'.") ekey = (params['event_type'].lower().strip(),) if ekey not in EventType: raise web.HTTPBadRequest(text=f"Unknown event type '{params['event_type']}'") event_type = EventType(ekey) req_fields = { 'user_name', 'occurred_at', 'event_type', } other_fields = { 'document_id', 'document', 'user_id', 'user', } if 'user_id' not in params and 'user' not in params: raise web.HTTPBadRequest(text="One of 'user_id' or 'user' must be supplied to create Event.") match event_type: case EventType.PLAIN: req_fields.add('message') case EventType.SUBSCRIBER: req_fields.add('tier') req_fields.add('subscribed_length') other_fields.add('message') case EventType.CHEER: req_fields.add('amount') other_fields.add('cheer_type') other_fields.add('message') case EventType.RAID: req_fields.add('viewer_count') create_fields = req_fields.union(other_fields) 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 {event_type} event creation.") if missing := next((key for key in req_fields if key not in params), None): raise web.HTTPBadRequest(text=f"{event_type} Event params missing required key '{missing}'") @classmethod async def create(cls, app: web.Application, **kwargs): data = app[datamodelsv] # EventD = data.EventDetails ekey = (kwargs['event_type'].lower().strip(),) if ekey not in EventType: raise web.HTTPBadRequest(text=f"Unknown event type '{kwargs['event_type']}'") event_type = EventType(ekey) params = {} typparams = {} match event_type: case EventType.PLAIN: typtab = data.plain_events typparams['message'] = kwargs['message'] case EventType.CHEER: typtab = data.cheer_events typparams['amount'] = kwargs['amount'] typparams['cheer_type'] = kwargs.get('cheer_type') typparams['message'] = kwargs.get('message') case EventType.RAID: typtab = data.raid_events typparams['viewer_count'] = kwargs.get('viewer_count') case EventType.SUBSCRIBER: typtab = data.subscriber_events typparams['tier'] = kwargs['tier'] typparams['subscribed_length'] = kwargs['subscribed_length'] typparams['message'] = kwargs.get('message') case _: raise ValueError("Invalid EventType") # TODO: This really really should be a transaction # Create Document if required if 'document' in kwargs: from .documents import Document doc_args = kwargs['document'] await Document.validate_create_params(doc_args) doc = await Document.create(app, **doc_args) document_id = doc.row.document_id params['document_id'] = document_id elif 'document_id' in kwargs: document_id = kwargs['document_id'] params['document_id'] = document_id # Create User if required if 'user' in kwargs: from .users import User user_args = kwargs['user'] await User.validate_create_params(user_args) user = await User.create(app, **user_args) user_id = user.row.user_id if 'user_id' in kwargs and not kwargs['user_id'] == user_id: raise web.HTTPBadRequest(text="Provided 'user_id' does not match provided 'user'.") else: user_id = kwargs['user_id'] params['user_id'] = user_id # Create Event row params['event_type'] = event_type params['user_name'] = kwargs['user_name'] params['occurred_at'] = datetime.fromisoformat(kwargs['occurred_at']) eventrow = await data.Events.create(**params) typparams['event_id'] = eventrow.event_id # Create Event type row typrow = await typtab.insert(**typparams) details = await data.EventDetails.fetch(eventrow.event_id) assert details is not None return cls(app, details) async def edit(self, **kwargs): data = self.data # EventD = data.EventDetails if 'event_type' in kwargs: raise web.HTTPBadRequest(text="You cannot change the type of an event after creation.") typparams = {} match self.row.event_type: case EventType.PLAIN: typtab = data.plain_events if 'message' in kwargs: typparams['message'] = kwargs['message'] case EventType.CHEER: typtab = data.cheer_events for key in ('amount', 'cheer_type', 'message'): if key in kwargs: typparams[key] = kwargs[key] case EventType.RAID: typtab = data.raid_events for key in ('viewer_count',): if key in kwargs: typparams[key] = kwargs[key] case EventType.SUBSCRIBER: typtab = data.subscriber_events for key in ('tier', 'subscribed_length', 'message'): if key in kwargs: typparams[key] = kwargs[key] if typparams: await typtab.update_where(event_id=self.row.event_id).set(**typparams) await self.row.refresh() async def delete(self): payload = await self.prepare() if self.row.document_id: await self.data.Document.table.delete_where(document_id=self.row.document_id) await self.data.Events.table.delete_where(event_id=self.row.event_id) await self.row.refresh() return payload async def get_user(self): from .users import User return await User.fetch_from_id(self.app, self.row.user_id) async def get_document(self): from .documents import Document if self.row.document_id: return await Document.fetch_from_id(self.app, self.row.document_id) async def prepare(self): row = await self.row.refresh() assert row is not None data = self.data user = await self.get_user() assert user is not None document = await self.get_document() payload = { 'event_id': self.row.event_id, 'document_id': self.row.document_id, 'document': await document.prepare() if document else None, 'user_id': self.row.user_id, 'user': await user.prepare(), 'user_name': self.row.user_name, 'occurred_at': self.row.occurred_at.isoformat(), 'created_at': self.row.created_at.isoformat(), 'event_type': self.row.event_type.value[0], } match row.event_type: case EventType.PLAIN: payload['message'] = row.plain_message case EventType.SUBSCRIBER: payload['tier'] = row.subscriber_tier payload['subscribed_length'] = row.subscriber_length payload['message'] = row.subscriber_message case EventType.CHEER: payload['amount'] = row.cheer_amount payload['cheer_type'] = row.cheer_type payload['message'] = row.cheer_message case EventType.RAID: payload['viewer_count'] = row.raid_visitor_count return payload @routes.view('/events') @routes.view('/events/', name='events') class EventsView(web.View): async def post(self): request = self.request params = await request.json() if 'user_id' in request: params.setdefault('user_id', request['user_id']) await Event.validate_create_params(params) logger.info(f"Creating a new event with args: {params=}") event = await Event.create(self.request.app, **params) logger.debug(f"Created event: {event!r}") payload = await event.prepare() return web.json_response(payload) async def get(self): request = self.request filter_params = {} keys = [ 'event_id', 'document_id', 'document_seal', 'user_id', 'user_name', 'occurred_before', 'occurred_after', 'created_before', 'created_after', 'event_type', ] for key in keys: value = request.query.get(key, request.get(key, None)) filter_params[key] = value logger.info(f"Querying events with params: {filter_params=}") events = await Event.query(request.app, **filter_params) payload = [await event.prepare() for event in events] return web.json_response(payload) @routes.view('/events/{event_id}') @routes.view('/events/{event_id}/', name='event') class EventView(web.View): async def resolve_event(self): request = self.request event_id = request.match_info['event_id'] event = await Event.fetch_from_id(request.app, int(event_id)) if event is None: raise web.HTTPNotFound(text="No event exists with the given ID.") return event async def get(self): event = await self.resolve_event() logger.info(f"Received GET for event {event=}") payload = await event.prepare() return web.json_response(payload) async def patch(self): event = await self.resolve_event() params = await self.request.json() edit_data = {} edit_fields = {'message', 'amount', 'cheer_type', 'viewer_count', 'tier', 'subscriber_length', 'message'} for key, value in params.items(): if key not in edit_fields: raise web.HTTPBadRequest(text=f"You cannot update field '{key}' of User!") edit_data[key] = value for key in edit_fields: if key in self.request: edit_data.setdefault(key, self.request[key]) logger.info(f"Received PATCH for event {event} with params: {params}") await event.edit(**edit_data) payload = await event.prepare() return web.json_response(payload) async def delete(self): event = await self.resolve_event() logger.info(f"Received DELETE for event {event}") payload = await event.delete() return web.json_response(payload) @routes.route('*', "/events/{event_id}/user") @routes.route('*', "/events/{event_id}/user{tail:/.*}") async def event_user_route(request: web.Request): event_id = int(request.match_info['event_id']) event = await Event.fetch_from_id(request.app, event_id) if event is None: raise web.HTTPNotFound(text="No event exists with the given ID.") tail = request.match_info.get('tail', '') new_path = "/users/{user_id}".format(user_id=event.row.user_id) + tail logger.info(f"Redirecting {request=} to {new_path}") new_request = request.clone(rel_url=new_path) new_request['user_id'] = event.row.user_id match_info = await request.app.router.resolve(new_request) new_request._match_info = match_info match_info.current_app = request.app if match_info.handler: return await match_info.handler(new_request) else: logger.info(f"Could not find handler matching {new_request}") raise web.HTTPNotFound() @routes.route('*', "/events/{event_id}/document") @routes.route('*', "/events/{event_id}/document{tail:/.*}") async def event_document_route(request: web.Request): event_id = int(request.match_info['event_id']) event = await Event.fetch_from_id(request.app, event_id) if event is None: raise web.HTTPNotFound(text="No event exists with the given ID.") tail = request.match_info.get('tail', '') document = await event.get_document() if document is None: if request.method == 'POST' and not tail: new_path = '/documents' logger.info(f"Redirecting {request=} to POST /documents") new_request = request.clone(rel_url=new_path) new_request['event_id'] = event_id match_info = await request.app.router.resolve(new_request) new_request._match_info = match_info match_info.current_app = request.app return await match_info.handler(new_request) else: raise web.HTTPNotFound(text="This event has no document.") else: document_id = document.row.document_id # Redirect to POST /documents/{document_id}/... new_path = f"/documents/{document_id}".format(document_id=document_id) + tail logger.info(f"Redirecting {request=} to {new_path}") new_request = request.clone(rel_url=new_path) new_request['event_id'] = event_id new_request['document_id'] = document_id match_info = await request.app.router.resolve(new_request) new_request._match_info = match_info match_info.current_app = request.app if match_info.handler: return await match_info.handler(new_request) else: logger.info(f"Could not find handler matching {new_request}") raise web.HTTPNotFound()