From: Michael Tremer <michael.tremer@ipfire.org>
Date: Wed, 2 Jul 2025 13:43:51 +0000 (+0000)
Subject: mirrors: Migrate to SQLModel
X-Git-Url: http://git.ipfire.org/cgi-bin/gitweb.cgi?a=commitdiff_plain;h=5950fdbbb049ea6338c652f964a68d5b9006b00c;p=pbs.git

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 e3b3ad5d..bcf8c325 100644
--- 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