Convert models to new ORM style

Some queries have been updated, but not all.

meshinfo/collector.py

@@ -291,6 +291,7 @@ def save_nodes(
     """
     if count is None:
         count = defaultdict(int)
+    timestamp = pendulum.now()
     node_models = []
     for node in nodes:
         count["nodes: total"] += 1
@@ -311,7 +312,7 @@ def save_nodes(
                 logger.debug("Updated node in database", model=model)
             node_models.append(model)
 
-            model.last_seen = pendulum.now()
+            model.last_seen = timestamp
             model.status = NodeStatus.ACTIVE
 
             for model_attr, node_attr in MODEL_TO_SYSINFO_ATTRS.items():
@@ -393,6 +394,7 @@ def save_links(
         for node in dbsession.query(Node).filter(Node.status == NodeStatus.ACTIVE)
     }
 
+    timestamp = pendulum.now()
     link_models = []
     for link in links:
         count["links: total"] += 1
@@ -425,7 +427,7 @@ def save_links(
         link_models.append(model)
 
         model.status = LinkStatus.CURRENT
-        model.last_seen = pendulum.now()
+        model.last_seen = timestamp
 
         for attribute in [
             "type",

meshinfo/historical.py

@@ -112,9 +112,7 @@ def update_node_stats(self, node: Node) -> bool:
                     node.link_count,
                     len(node.services),
                     node.up_time_seconds,
-                    node.load_averages[0]
-                    if isinstance(node.load_averages, list)
-                    else None,
+                    node.load_averages[0] if node.load_averages is not None else None,
                     node.radio_link_count,
                     node.dtd_link_count,
                     node.tunnel_link_count,

meshinfo/models/collector.py

@@ -1,42 +1,44 @@
+from __future__ import annotations
+
 import pendulum
-import sqlalchemy as sa
-from sqlalchemy.orm import relationship
+from sqlalchemy import JSON, ForeignKey, String
+from sqlalchemy.orm import Mapped, mapped_column, relationship
 
 from ..poller import PollingError
-from .meta import Base, PDateTime
+from .meta import Base
+
+
+class NodeError(Base):
+    """Information about nodes with errors during collection."""
+
+    __tablename__ = "node_error"
+
+    timestamp: Mapped[pendulum.DateTime] = mapped_column(
+        ForeignKey("collector_stat.started_at"), primary_key=True
+    )
+    ip_address: Mapped[str] = mapped_column(String(15), primary_key=True)
+    dns_name: Mapped[str] = mapped_column(String(70))
+    error_type: Mapped[PollingError]
+    details: Mapped[str]
 
 
 class CollectorStat(Base):
     """Statistics from the collection process."""
 
     __tablename__ = "collector_stat"
 
-    started_at = sa.Column(PDateTime(), primary_key=True)
-    finished_at = sa.Column(PDateTime(), default=pendulum.now, nullable=False)
-    node_count = sa.Column(sa.Integer, nullable=False)
-    link_count = sa.Column(sa.Integer, nullable=False)
-    error_count = sa.Column(sa.Integer, nullable=False)
-    polling_duration = sa.Column(sa.Float, nullable=False)
-    total_duration = sa.Column(sa.Float, nullable=False)
-    other_stats = sa.Column(sa.JSON, nullable=False)
-
-    node_errors = relationship(
-        "NodeError", foreign_keys="NodeError.timestamp", cascade="all, delete-orphan"
+    started_at: Mapped[pendulum.DateTime] = mapped_column(primary_key=True)
+    finished_at: Mapped[pendulum.DateTime] = mapped_column(default=pendulum.now)
+    node_count: Mapped[int]
+    link_count: Mapped[int]
+    error_count: Mapped[int]
+    polling_duration: Mapped[float]
+    total_duration: Mapped[float]
+    other_stats: Mapped[dict] = mapped_column(JSON)
+
+    node_errors: Mapped[list[NodeError]] = relationship(
+        foreign_keys="NodeError.timestamp", cascade="all, delete-orphan"
     )
 
     def __repr__(self):
         return f"<models.CollectorStat(started_at={self.started_at})>"
-
-
-class NodeError(Base):
-    """Information about nodes with errors during collection."""
-
-    __tablename__ = "node_error"
-
-    timestamp = sa.Column(
-        PDateTime(), sa.ForeignKey("collector_stat.started_at"), primary_key=True
-    )
-    ip_address = sa.Column(sa.String(15), primary_key=True)
-    dns_name = sa.Column(sa.String(70), nullable=False)
-    error_type = sa.Column(sa.Enum(PollingError, native_enum=False), nullable=False)
-    details = sa.Column(sa.UnicodeText, nullable=False)

meshinfo/models/link.py

@@ -1,45 +1,51 @@
+"""Database model(s) for representing links between nodes."""
+
+from typing import TYPE_CHECKING, Optional
+
 import pendulum
-import sqlalchemy as sa
-from sqlalchemy.orm import relationship
+from sqlalchemy import ForeignKey
+from sqlalchemy.orm import Mapped, mapped_column, relationship
 
 from ..types import LinkId, LinkStatus, LinkType
-from .meta import Base, PDateTime
+from .meta import Base
+
+if TYPE_CHECKING:
+    from .node import Node
 
 
 class Link(Base):
     """Represents a link between two nodes."""
 
     __tablename__ = "link"
 
-    source_id = sa.Column(sa.Integer, sa.ForeignKey("node.node_id"), primary_key=True)
-    destination_id = sa.Column(
-        sa.Integer, sa.ForeignKey("node.node_id"), primary_key=True
+    source_id: Mapped[int] = mapped_column(ForeignKey("node.node_id"), primary_key=True)
+    destination_id: Mapped[int] = mapped_column(
+        ForeignKey("node.node_id"), primary_key=True
     )
-    type = sa.Column(sa.Enum(LinkType, native_enum=False), primary_key=True)
-    status = sa.Column(sa.Enum(LinkStatus, native_enum=False), nullable=False)
-    last_seen = sa.Column(PDateTime(), nullable=False, default=pendulum.now)
-
-    olsr_cost = sa.Column(sa.Float)
-    distance = sa.Column(sa.Float)
-    bearing = sa.Column(sa.Float)
-
-    signal = sa.Column(sa.Float)
-    noise = sa.Column(sa.Float)
-    tx_rate = sa.Column(sa.Float)
-    rx_rate = sa.Column(sa.Float)
-    quality = sa.Column(sa.Float)
-    neighbor_quality = sa.Column(sa.Float)
-
-    created_at = sa.Column(PDateTime(), default=pendulum.now, nullable=False)
-    last_updated_at = sa.Column(
-        PDateTime(),
-        default=pendulum.now,
-        onupdate=pendulum.now,
-        nullable=False,
+    type: Mapped[LinkType] = mapped_column(primary_key=True)
+    status: Mapped[LinkStatus]
+    last_seen: Mapped[pendulum.DateTime] = mapped_column(default=pendulum.now)
+
+    olsr_cost: Mapped[float]
+    distance: Mapped[Optional[float]]
+    bearing: Mapped[Optional[float]]
+
+    signal: Mapped[Optional[float]]
+    noise: Mapped[Optional[float]]
+    tx_rate: Mapped[Optional[float]]
+    rx_rate: Mapped[Optional[float]]
+    quality: Mapped[Optional[float]]
+    neighbor_quality: Mapped[Optional[float]]
+
+    created_at: Mapped[pendulum.DateTime] = mapped_column(default=pendulum.now)
+    last_updated_at: Mapped[pendulum.DateTime] = mapped_column(
+        default=pendulum.now, onupdate=pendulum.now
     )
 
-    source = relationship("Node", foreign_keys="Link.source_id", back_populates="links")
-    destination = relationship("Node", foreign_keys="Link.destination_id")
+    source: Mapped["Node"] = relationship(
+        foreign_keys="Link.source_id", back_populates="links"
+    )
+    destination: Mapped["Node"] = relationship(foreign_keys="Link.destination_id")
 
     @property
     def signal_noise_ratio(self):

meshinfo/models/meta.py

@@ -1,7 +1,7 @@
 from __future__ import annotations
 
 import pendulum
-from sqlalchemy.orm import declarative_base
+from sqlalchemy.orm import DeclarativeBase
 from sqlalchemy.schema import MetaData
 from sqlalchemy.types import TIMESTAMP, TypeDecorator
 
@@ -17,14 +17,13 @@
 }
 
 metadata = MetaData(naming_convention=NAMING_CONVENTION)
-Base = declarative_base(metadata=metadata)
 
 
 class PDateTime(TypeDecorator):
     """SQLAlchemy type to wrap `pendulum.datetime` instead of `datetime.datetime`."""
 
     impl = TIMESTAMP(timezone=True)
-    cache_ok = False
+    cache_ok = True
 
     def process_bind_param(self, value, dialect):
         if value is not None:
@@ -37,5 +36,12 @@ def process_bind_param(self, value, dialect):
     def process_result_value(self, value, dialect):
         if value is not None:
             value = pendulum.instance(value)
-
         return value
+
+
+class Base(DeclarativeBase):
+    metadata = metadata
+
+    type_annotation_map = {
+        pendulum.DateTime: PDateTime,
+    }

meshinfo/models/node.py

@@ -1,84 +1,77 @@
+from typing import TYPE_CHECKING, Optional
+
 import pendulum
-from sqlalchemy import (
-    JSON,
-    Boolean,
-    Column,
-    Enum,
-    Float,
-    Index,
-    Integer,
-    String,
-    Unicode,
-)
-from sqlalchemy.orm import relationship
+from sqlalchemy import JSON, Enum, Index, String, Unicode
+from sqlalchemy.orm import Mapped, mapped_column, relationship
 
 from ..types import Band, NodeStatus
-from .meta import Base, PDateTime
+from .meta import Base
+
+if TYPE_CHECKING:
+    from .link import Link
 
 
 class Node(Base):
     """Information about a node in the mesh network."""
 
     __tablename__ = "node"
 
-    id = Column("node_id", Integer, primary_key=True)
-    name = Column(String(70), nullable=False)
-    status = Column(Enum(NodeStatus, native_enum=False), nullable=False)
-    display_name = Column(String(70), nullable=False)
+    id: Mapped[int] = mapped_column("node_id", primary_key=True)
+    name: Mapped[str] = mapped_column(String(70))
+    status: Mapped[NodeStatus]
+    display_name: Mapped[str] = mapped_column(String(70))
 
     # store the wireless/primary IP address
-    ip_address = Column("wlan_ip", String(15), nullable=False)
-    description = Column(Unicode(1024), nullable=False)
+    ip_address: Mapped[str] = mapped_column("wlan_ip", String(15))
+    description: Mapped[str] = mapped_column(Unicode(1024))
 
     # store the MAC address (without colons) corresponding the primary interface
-    mac_address = Column("wlan_mac_address", String(12), nullable=False)
-
-    last_seen = Column(PDateTime(), nullable=False)
-
-    up_time = Column(String(25), nullable=False)
-    up_time_seconds = Column(Integer)
-    load_averages = Column(JSON())
-    model = Column(String(50), nullable=False)
-    board_id = Column(String(50), nullable=False)
-    firmware_version = Column(String(50), nullable=False)
-    firmware_manufacturer = Column(String(100), nullable=False)
-    api_version = Column(String(5), nullable=False)
-
-    latitude = Column(Float)
-    longitude = Column(Float)
-    grid_square = Column(String(20), nullable=False)
-
-    ssid = Column(String(50), nullable=False)
-    channel = Column(String(50), nullable=False)
-    channel_bandwidth = Column(String(50), nullable=False)
-    band = Column(
+    mac_address: Mapped[str] = mapped_column("wlan_mac_address", String(12))
+
+    last_seen: Mapped[pendulum.DateTime]
+
+    up_time: Mapped[str] = mapped_column(String(25))
+    up_time_seconds: Mapped[Optional[int]]
+    load_averages: Mapped[Optional[list[float]]] = mapped_column(JSON)
+    model: Mapped[str] = mapped_column(String(50))
+    board_id: Mapped[str] = mapped_column(String(50))
+    firmware_version: Mapped[str] = mapped_column(String(50))
+    firmware_manufacturer: Mapped[str] = mapped_column(String(100))
+    api_version: Mapped[str] = mapped_column(String(5))
+
+    latitude: Mapped[Optional[float]]
+    longitude: Mapped[Optional[float]]
+    grid_square: Mapped[str] = mapped_column(String(20))
+
+    ssid: Mapped[str] = mapped_column(String(50))
+    channel: Mapped[str] = mapped_column(String(50))
+    channel_bandwidth: Mapped[str] = mapped_column(String(50))
+    band: Mapped[Band] = mapped_column(
         Enum(Band, values_callable=lambda x: [e.value for e in x], native_enum=False),
-        nullable=False,
     )
 
-    services = Column(JSON(), nullable=False)
+    services: Mapped[dict] = mapped_column(JSON)
 
     # As of API v1.10 this is irrelevant (because it is always enabled)
     # (probably worth deleting at some point in the future)
-    tunnel_installed = Column(Boolean(), nullable=False, default=True)
-    active_tunnel_count = Column(Integer(), nullable=False)
-
-    link_count = Column(Integer())
-    radio_link_count = Column(Integer())
-    dtd_link_count = Column(Integer())
-    tunnel_link_count = Column(Integer())
-
-    system_info = Column(JSON(), nullable=False)
-
-    created_at = Column(PDateTime(), default=pendulum.now, nullable=False)
-    last_updated_at = Column(
-        PDateTime(),
-        default=pendulum.now,
-        onupdate=pendulum.now,
-        nullable=False,
+    tunnel_installed: Mapped[bool] = mapped_column(default=True)
+    active_tunnel_count: Mapped[int]
+
+    link_count: Mapped[Optional[int]]
+    radio_link_count: Mapped[Optional[int]]
+    dtd_link_count: Mapped[Optional[int]]
+    tunnel_link_count: Mapped[Optional[int]]
+
+    system_info: Mapped[dict] = mapped_column(JSON)
+
+    created_at: Mapped[pendulum.DateTime] = mapped_column(default=pendulum.now)
+    last_updated_at: Mapped[pendulum.DateTime] = mapped_column(
+        default=pendulum.now, onupdate=pendulum.now
     )
 
-    links = relationship("Link", foreign_keys="Link.source_id", back_populates="source")
+    links: Mapped["Link"] = relationship(
+        foreign_keys="Link.source_id", back_populates="source"
+    )
 
     # Is this premature optimization?
     Index("idx_mac_name", mac_address, name)