kiwi-vpn/api/kiwi_vpn_api/db/device.py

"""
Python representation of `device` table.
"""

from __future__ import annotations

from datetime import datetime
from typing import TYPE_CHECKING

from sqlalchemy.exc import IntegrityError
from sqlmodel import Field, Relationship, SQLModel, UniqueConstraint

from .connection import Connection

if TYPE_CHECKING:
    from .user import User


class DeviceBase(SQLModel):
    """
    Common to all representations of devices
    """

    name: str
    type: str
    expiry: datetime | None


class DeviceCreate(DeviceBase):
    """
    Representation of a newly created device
    """


class DeviceRead(DeviceBase):
    """
    Representation of a device read via the API
    """

    approved: bool
    owner_name: str | None


class Device(DeviceBase, table=True):
    """
    Representation of `device` table
    """

    __table_args__ = (UniqueConstraint(
        "owner_name",
        "name",
    ),)

    id: int | None = Field(primary_key=True)
    approved: bool = Field(default=False)
    owner_name: str | None = Field(foreign_key="user.name")

    # no idea, but "User" (in quotes) doesn't work here
    # might be a future problem?
    owner: User = Relationship(
        back_populates="devices",
    )

    @classmethod
    def create(
        cls,
        *,
        owner: User,
        device: DeviceCreate,
    ) -> Device | None:
        """
        Create a new device in the database.
        """

        try:
            with Connection.session as db:
                new_device = cls.from_orm(device)
                new_device.owner = owner

                db.add(new_device)
                db.commit()
                db.refresh(new_device)

                return new_device

        except IntegrityError:
            # device already existed
            return None

    @classmethod
    def get(cls, id: int) -> Device | None:
        """
        Load device from database by id.
        """

        with Connection.session as db:
            return db.get(cls, id)

    def update(self) -> None:
        """
        Update this device in the database.
        """

        with Connection.session as db:
            db.add(self)
            db.commit()
            db.refresh(self)

    def delete(self) -> bool:
        """
        Delete this device from the database.
        """

        with Connection.session as db:
            db.delete(self)
            db.commit()