mirror of
https://github.com/nihilvux/bancho.py.git
synced 2026-09-17 15:33:06 -07:00
71 lines
1.7 KiB
Python
71 lines
1.7 KiB
Python
from __future__ import annotations
|
|
|
|
from datetime import datetime
|
|
from typing import TypedDict
|
|
from typing import cast
|
|
|
|
from sqlalchemy import Column
|
|
from sqlalchemy import DateTime
|
|
from sqlalchemy import Integer
|
|
from sqlalchemy import String
|
|
from sqlalchemy import func
|
|
from sqlalchemy import insert
|
|
from sqlalchemy import select
|
|
|
|
import app.state.services
|
|
from app.repositories import Base
|
|
|
|
|
|
class LogTable(Base):
|
|
__tablename__ = "logs"
|
|
|
|
id = Column("id", Integer, nullable=False, primary_key=True, autoincrement=True)
|
|
_from = Column("from", Integer, nullable=False)
|
|
to = Column("to", Integer, nullable=False)
|
|
action = Column("action", String(32), nullable=False)
|
|
msg = Column("msg", String(2048, collation="utf8"), nullable=True)
|
|
time = Column("time", DateTime, nullable=False, onupdate=func.now())
|
|
|
|
|
|
READ_PARAMS = (
|
|
LogTable.id,
|
|
LogTable._from.label("from"),
|
|
LogTable.to,
|
|
LogTable.action,
|
|
LogTable.msg,
|
|
LogTable.time,
|
|
)
|
|
|
|
|
|
class Log(TypedDict):
|
|
id: int
|
|
_from: int
|
|
to: int
|
|
action: str
|
|
msg: str | None
|
|
time: datetime
|
|
|
|
|
|
async def create(
|
|
_from: int,
|
|
to: int,
|
|
action: str,
|
|
msg: str,
|
|
) -> Log:
|
|
"""Create a new log entry in the database."""
|
|
insert_stmt = insert(LogTable).values(
|
|
{
|
|
"from": _from,
|
|
"to": to,
|
|
"action": action,
|
|
"msg": msg,
|
|
"time": func.now(),
|
|
},
|
|
)
|
|
rec_id = await app.state.services.database.execute(insert_stmt)
|
|
|
|
select_stmt = select(*READ_PARAMS).where(LogTable.id == rec_id)
|
|
log = await app.state.services.database.fetch_one(select_stmt)
|
|
assert log is not None
|
|
return cast(Log, log)
|