mirrors: Migrate to SQLModel

Signed-off-by: Michael Tremer <michael.tremer@ipfire.org>

diff --git a/src/buildservice/mirrors.py b/src/buildservice/mirrors.py
index e3b3ad5db5b7c562bf53422a0f704dc5a0f3e3b4..bcf8c3259b72601d5a64587160a9744b4fd6d89a 100644 (file)
--- a/src/buildservice/mirrors.py
+++ b/src/buildservice/mirrors.py
@@ -7,14 +7,15 @@ import ipaddress
 import logging
 import os.path
 import random
+import sqlalchemy
+import sqlmodel
 import socket
 import tornado.netutil
 import urllib.parse
 
 import location
 
-import sqlalchemy
-from sqlalchemy import ARRAY, Boolean, Column, DateTime, Double, ForeignKey, Integer, Text
+# Import types
 from sqlalchemy.dialects.postgresql import INET
 
 from . import base
@@ -121,7 +122,7 @@ class Mirrors(base.Object):
                                )
 
 
-class Mirror(database.Base, database.BackendMixin, database.SoftDeleteMixin):
+class Mirror(sqlmodel.SQLModel, database.BackendMixin, database.SoftDeleteMixin, table=True):
        __tablename__ = "mirrors"
 
        def __str__(self):
@@ -135,17 +136,19 @@ class Mirror(database.Base, database.BackendMixin, database.SoftDeleteMixin):
 
        # ID
 
-       id = Column(Integer, primary_key=True)
+       id : int = sqlmodel.Field(primary_key=True, exclude=True)
 
        # Hostname
 
-       hostname = Column(Text, unique=True, nullable=False)
+       hostname: str = sqlmodel.Field(
+           sa_column=sqlalchemy.Column(sqlalchemy.Text, unique=True, nullable=False),
+       )
 
        # XXX Must be unique over non-deleted items
 
        # Path
 
-       path = Column(Text, nullable=False)
+       path: str
 
        # URL
 
@@ -170,41 +173,37 @@ class Mirror(database.Base, database.BackendMixin, database.SoftDeleteMixin):
 
        # Last Check Success - True if the last check was successful
 
-       last_check_success = Column(Boolean, nullable=False, default=False)
+       last_check_success: bool = sqlmodel.Field(default=False)
 
        # Last Check At
 
-       last_check_at = Column(DateTime(timezone=False))
+       last_check_at: datetime.datetime
 
        # Error Message when the check has been unsuccessful
 
-       error = Column(Text, nullable=False, default="")
+       error: str = sqlmodel.Field(default="")
 
        # Created At
 
-       created_at = Column(
-               DateTime(timezone=False), nullable=False, server_default=sqlalchemy.func.current_timestamp(),
+       created_at : datetime.datetime = sqlmodel.Field(
+               sa_column_kwargs = {"server_default" : sqlalchemy.text("CURRENT_TIMESTAMP")}
        )
 
        # Created By ID
 
-       created_by_id = Column(Integer, ForeignKey("users.id"), nullable=False)
+       created_by_id: int = sqlmodel.Field(foreign_key="users.id")
 
        # Created By
 
-       created_by = sqlalchemy.orm.relationship(
-               "User", foreign_keys=[created_by_id], lazy="joined", innerjoin=True,
-       )
+       created_by: "User" = sqlmodel.Relationship()
 
        # Deleted By ID
 
-       deleted_by_id = Column(Integer, ForeignKey("users.id"))
+       deleted_by_id: int | None = sqlmodel.Field(foreign_key="users.id")
 
        # Deleted By
 
-       deleted_by = sqlalchemy.orm.relationship(
-               "User", foreign_keys=[deleted_by_id], lazy="selectin",
-       )
+       deleted_by: "User" = sqlmodel.Relationship()
 
        def has_perm(self, user):
                # Anonymous users have no permission
@@ -216,27 +215,29 @@ class Mirror(database.Base, database.BackendMixin, database.SoftDeleteMixin):
 
        # Owner
 
-       owner = Column(Text, nullable=False)
+       owner: str
 
        # Contact
 
-       contact = Column(Text, nullable=False)
+       contact: str
 
        # Notes
 
-       notes = Column(Text, nullable=False, default="")
+       notes: str = sqlmodel.Field(default="")
 
        # Country Code
 
-       country_code = Column(Text)
+       country_code: str | None
 
        # ASN
 
-       asn = Column(Integer)
+       asn: int
 
        # Addresses IPv6
 
-       addresses_ipv6 = Column(ARRAY(INET), nullable=False, default=[])
+       addresses_ipv6: list[ipaddress.IPv6Address] = sqlmodel.Field(
+               sa_column=sqlalchemy.Column(sqlalchemy.ARRAY(INET), nullable=False, default=list)
+       )
 
        def supports_ipv6(self):
                """
@@ -249,7 +250,9 @@ class Mirror(database.Base, database.BackendMixin, database.SoftDeleteMixin):
 
        # Addresses IPv4
 
-       addresses_ipv4 = Column(ARRAY(INET), nullable=False, default=[])
+       addresses_ipv6: list[ipaddress.IPv4Address] = sqlmodel.Field(
+               sa_column=sqlalchemy.Column(sqlalchemy.ARRAY(INET), nullable=False, default=list)
+       )
 
        def supports_ipv4(self):
                """
@@ -579,7 +582,7 @@ class Mirror(database.Base, database.BackendMixin, database.SoftDeleteMixin):
                return serves_file
 
 
-class MirrorCheck(database.Base):
+class MirrorCheck(sqlmodel.SQLModel, table=True):
        """
                An object that represents a single mirror check
        """
@@ -587,29 +590,33 @@ class MirrorCheck(database.Base):
 
        # Mirror ID
 
-       mirror_id = Column(Integer, ForeignKey("mirrors.id"), primary_key=True, nullable=False)
+       mirror_id: int = sqlmodel.Field(primary_key=True, foreign_key="mirrors.id")
 
        # Checked At
 
-       checked_at = Column(DateTime(timezone=None), primary_key=True, nullable=False,
-               server_default=sqlalchemy.func.current_timestamp())
+       checked_at: datetime.datetime = sqlmodel.Field(
+               sa_column=sqlalchemy.Column(
+                       sqlalchemy.DateTime(timezone=False), primary_key=True, nullable=False,
+                       server_default=sqlalchemy.func.current_timestamp()
+               ),
+       )
 
        # Success
 
-       success = Column(Boolean, nullable=False)
+       success: bool
 
        # Response Time
 
-       response_time = Column(Double)
+       response_time: float
 
        # HTTP Status
 
-       http_status = Column(Integer)
+       http_status: int
 
        # Last Sync At
 
-       last_sync_at = Column(DateTime(timezone=None))
+       last_sync_at: datetime.datetime
 
        # Error
 
-       error = Column(Text)
+       error: str