Swap to using SQLModel instead of pony.orm

freyr/__init__.py

@@ -20,7 +20,7 @@
 
 from freyr.console import CONSOLE
 
-__version__ = "0.5.3"
+__version__ = "0.6.0"
 
 
 def get_cache() -> Path:

freyr/__main__.py

@@ -9,6 +9,7 @@
 
 from freyr import __version__, elapsed_timer, get_project, setup_logging
 from freyr.constants import constants
+from freyr.database import create_db_and_tables
 from freyr.routers.api import router as api_router
 from freyr.routers.html import router as html_router
 
@@ -30,6 +31,8 @@ def create_app() -> FastAPI:
 async def startup_event() -> None:
     setup_logging()
 
+    create_db_and_tables()
+
     LOGGER.info(
         "Listening on %s:%s", constants.settings.website.host, constants.settings.website.port
     )

freyr/database.py

@@ -0,0 +1,18 @@
+from sqlmodel import Session, SQLModel, create_engine
+
+from freyr.constants import constants
+
+sqlite_file_name = "database.db"
+sqlite_url = f"sqlite:///{sqlite_file_name}"
+
+connect_args = {"check_same_thread": False}
+engine = create_engine(constants.settings.database.db_url, echo=False, connect_args=connect_args)
+
+
+def create_db_and_tables() -> None:
+    SQLModel.metadata.create_all(engine)
+
+
+def get_session() -> Session:
+    with Session(engine) as session:
+        yield session

freyr/models.py

@@ -0,0 +1,86 @@
+from datetime import datetime
+from decimal import Decimal
+from typing import Annotated, Self
+
+from sqlmodel import Field, Relationship, SQLModel
+
+
+class DeviceBase(SQLModel):
+    name: str = Field(index=True)
+
+
+class Device(DeviceBase, table=True):
+    __tablename__ = "devices"
+
+    id: int | None = Field(default=None, primary_key=True)
+    readings: list["Reading"] = Relationship(back_populates="device")
+
+
+class DeviceCreate(DeviceBase):
+    pass
+
+
+class DevicePublic(DeviceBase):
+    id: int
+    readings: list["ReadingPublic"] = Field(default_factory=list)
+
+
+class ReadingBase(SQLModel):
+    temperature: Decimal | None = None
+    humidity: Decimal | None = None
+
+
+class Reading(ReadingBase, table=True):
+    __tablename__ = "readings"
+
+    id: int | None = Field(default=None, primary_key=True)
+    timestamp: datetime
+    device_id: int = Field(foreign_key="devices.id")
+    device: Device = Relationship(back_populates="readings")
+
+    def __lt__(self: Self, other) -> int:  # noqa: ANN001
+        if not isinstance(other, Reading):
+            raise NotImplementedError
+        return self.timestamp < other.timestamp
+
+    def __eq__(self: Self, other) -> bool:  # noqa: ANN001
+        if not isinstance(other, Reading):
+            raise NotImplementedError
+        return self.timestamp == other.timestamp
+
+    def __hash__(self: Self) -> int:
+        return hash((type(self), self.timestamp))
+
+
+class ReadingCreate(ReadingBase):
+    device_id: int | None = None
+    timestamp: datetime | None = None
+
+
+class ReadingPublic(ReadingBase):
+    id: int
+    timestamp: datetime
+
+
+class Summary(SQLModel):
+    class Reading(SQLModel):
+        timestamp: datetime
+        temperature: Annotated[Decimal, Field(decimal_places=2)] | None = None
+        humidity: Annotated[Decimal, Field(decimal_places=2)] | None = None
+
+        def __lt__(self: Self, other) -> int:  # noqa: ANN001
+            if not isinstance(other, Summary.Reading):
+                raise NotImplementedError
+            return self.timestamp < other.timestamp
+
+        def __eq__(self: Self, other) -> bool:  # noqa: ANN001
+            if not isinstance(other, Summary.Reading):
+                raise NotImplementedError
+            return self.timestamp == other.timestamp
+
+        def __hash__(self: Self) -> int:
+            return hash((type(self), self.timestamp))
+
+    highs: list[Reading] = Field(default_factory=list)
+    averages: list[Reading] = Field(default_factory=list)
+    lows: list[Reading] = Field(default_factory=list)