"""DB models for the project."""
import datetime
import functools
import typing
import sqlalchemy
import sqlalchemy.orm
from sqlalchemy import sql
from sqlalchemy.ext import asyncio as sqlalchemy_asyncio
from structlog import stdlib as structlog
from pinger_bot import config
from pinger_bot.config import gettext as _
log = structlog.get_logger()
[docs]Base = sqlalchemy.orm.declarative_base()
"""``Base`` object for all models.
.. seealso:: `<https://docs.sqlalchemy.org/en/14/orm/mapping_api.html#sqlalchemy.orm.declarative_base>`_
"""
[docs]class Server(Base):
"""A server model. Used to store the server, that was added to bot."""
[docs] __tablename__ = "pb_servers"
[docs] id: int = sqlalchemy.Column(sqlalchemy.Integer, sqlalchemy.Identity(), primary_key=True)
"""Unique ID of the server, primary key."""
[docs] host: str = sqlalchemy.Column(sqlalchemy.String(256), nullable=False)
"""Hostname of the server, can be number IP or domain. Always without port."""
[docs] port: int = sqlalchemy.Column(sqlalchemy.Integer, nullable=False)
"""Port of the server."""
[docs] max: int = sqlalchemy.Column(sqlalchemy.Integer, nullable=False)
"""Max players on the server."""
[docs] alias: typing.Optional[str] = sqlalchemy.Column(sqlalchemy.String(256), unique=True)
"""Alias of the server. Can be used as IP of the server."""
[docs] owner: int = sqlalchemy.Column(sqlalchemy.BigInteger, nullable=False)
"""Owner's Discord ID of the server."""
[docs] __table_args__ = (sqlalchemy.UniqueConstraint("host", "port", name="ip"),)
"""Unique constraint for host and port."""
[docs]class Ping(Base):
"""Represents a single ping record in DB."""
[docs] __tablename__ = "pb_pings"
[docs] id: int = sqlalchemy.Column(sqlalchemy.Integer, sqlalchemy.Identity(), primary_key=True)
"""Unique ID of the ping, primary key."""
[docs] host: str = sqlalchemy.Column(sqlalchemy.String(256), nullable=False)
"""Host of the server, for which ping was made. This is a foreign key constraint."""
[docs] port: int = sqlalchemy.Column(sqlalchemy.Integer, nullable=False)
"""Port of the server, for which ping was made. This is a foreign key constraint."""
[docs] time: datetime.datetime = sqlalchemy.Column( # skipcq: PYL-E1102
sqlalchemy.DateTime, server_default=sql.func.now(), nullable=False
)
"""Time of the ping. Default to current time."""
[docs] players: int = sqlalchemy.Column(sqlalchemy.Integer, nullable=False)
"""Players online in moment of the ping."""
[docs] __mapper_args__ = {"eager_defaults": True}
"""Some magic to make eager loading work.
.. seealso:: `<https://docs.sqlalchemy.org/en/14/orm/extensions/asyncio.html#synopsis-orm>`_
"""
[docs] __table_args__ = (
sqlalchemy.ForeignKeyConstraint(
["host", "port"], ["pb_servers.host", "pb_servers.port"], name="ping_to_server"
),
)
"""Define Foreign Key Constraint to associate :attr:`pb_pings.host <.Ping.host>` and \
:attr:`pb_pings.port <.Server.port>` to :attr:`pb_pings.host <.Ping.host>` and \
:attr:`pb_pings.port <.Server.port>`."""
[docs]class Database:
"""Some cached info about database."""
@functools.cached_property
[docs] def engine(self) -> sqlalchemy_asyncio.AsyncEngine:
"""Async engine of the Database."""
log.info(_("Starting DB..."))
return sqlalchemy_asyncio.create_async_engine(config.config.db_uri, echo=config.config.debug)
@functools.cached_property
[docs] def session(self) -> typing.Callable[[], typing.AsyncContextManager[sqlalchemy_asyncio.AsyncSession]]:
"""Async session of the Database."""
return sqlalchemy.orm.sessionmaker(self.engine, expire_on_commit=False, class_=sqlalchemy_asyncio.AsyncSession)
"""Initialized :py:class:`.Database` object."""