114 lines
2.3 KiB
Python
114 lines
2.3 KiB
Python
"""
|
|
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
|
|
"""
|
|
|
|
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)
|
|
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()
|