diff --git a/alws/database.py b/alws/database.py index a7a56c54e..bb612d8a5 100644 --- a/alws/database.py +++ b/alws/database.py @@ -1,10 +1,13 @@ # -*- mode:python; coding:utf-8; -*- # author: Vyacheslav Potoropin # created: 2021-06-22 -from sqlalchemy import create_engine -from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine -from sqlalchemy.ext.declarative import declarative_base -from sqlalchemy.orm import scoped_session, sessionmaker +from sqlalchemy import MetaData, create_engine +from sqlalchemy.ext.asyncio import ( + AsyncAttrs, + AsyncSession, + create_async_engine, +) +from sqlalchemy.orm import DeclarativeBase, scoped_session, sessionmaker from sqlalchemy.pool import NullPool from alws.config import settings @@ -18,12 +21,22 @@ sync_engine = create_engine( settings.sync_database_url, pool_pre_ping=True, pool_recycle=3600 ) -Base = declarative_base() + + +class Base(AsyncAttrs, DeclarativeBase): + __allow_unmapped__ = True + metadata = MetaData() + + sync_session_factory = sessionmaker(sync_engine, expire_on_commit=False) Session = sessionmaker(engine, expire_on_commit=False, class_=AsyncSession) SyncSession = scoped_session(sync_session_factory) -PulpBase = declarative_base() + +class PulpBase(AsyncAttrs, DeclarativeBase): + __allow_unmapped__ = True + + pulp_engine = create_engine( settings.pulp_database_url, pool_pre_ping=True, pool_recycle=3600 ) diff --git a/alws/models.py b/alws/models.py index afe2f67a4..f464b9226 100644 --- a/alws/models.py +++ b/alws/models.py @@ -1,7 +1,7 @@ import asyncio import datetime import re -from typing import Dict, List +from typing import Any, Dict, List, Literal, Optional import sqlalchemy from fastapi_users.db import ( @@ -12,8 +12,14 @@ SQLAlchemyBaseAccessTokenTable, ) from sqlalchemy.dialects.postgresql import JSONB -from sqlalchemy.ext.associationproxy import association_proxy -from sqlalchemy.orm import declarative_mixin, declared_attr, relationship +from sqlalchemy.ext.associationproxy import AssociationProxy, association_proxy +from sqlalchemy.orm import ( + Mapped, + declarative_mixin, + declared_attr, + mapped_column, + relationship, +) from sqlalchemy.sql import func from alws.constants import ( @@ -49,9 +55,9 @@ @declarative_mixin class TeamMixin: @declared_attr - def team_id(cls): + def team_id(cls) -> Mapped[Optional[int]]: # FIXME: Change nullable to False after owner population - return sqlalchemy.Column( + return mapped_column( sqlalchemy.Integer, sqlalchemy.ForeignKey( "teams.id", @@ -61,16 +67,16 @@ def team_id(cls): ) @declared_attr - def team(cls): + def team(cls) -> Mapped["Team"]: return relationship("Team") @declarative_mixin class PermissionsMixin: @declared_attr - def owner_id(cls): + def owner_id(cls) -> Mapped[Optional[int]]: # FIXME: Change nullable to False after owner population - return sqlalchemy.Column( + return mapped_column( sqlalchemy.Integer, sqlalchemy.ForeignKey( "users.id", @@ -80,10 +86,10 @@ def owner_id(cls): ) @declared_attr - def owner(cls): + def owner(cls) -> Mapped["User"]: return relationship("User") - permissions = sqlalchemy.Column( + permissions: Mapped[int] = mapped_column( sqlalchemy.Integer, nullable=False, default=764 ) @@ -120,12 +126,12 @@ def validate_permissions(permissions: int): @declarative_mixin class TimeMixin: @declared_attr - def started_at(cls): - return sqlalchemy.Column(sqlalchemy.DateTime, nullable=True) + def started_at(cls) -> Mapped[Optional[datetime.datetime]]: + return mapped_column(sqlalchemy.DateTime, nullable=True) @declared_attr - def finished_at(cls): - return sqlalchemy.Column(sqlalchemy.DateTime, nullable=True) + def finished_at(cls) -> Mapped[Optional[datetime.datetime]]: + return mapped_column(sqlalchemy.DateTime, nullable=True) PlatformRepo = sqlalchemy.Table( @@ -196,7 +202,6 @@ def finished_at(cls): ), ) - PlatformRoleMapping = sqlalchemy.Table( "platform_role_mapping", Base.metadata, @@ -224,35 +229,57 @@ def finished_at(cls): class Platform(PermissionsMixin, Base): __tablename__ = "platforms" - id = sqlalchemy.Column(sqlalchemy.Integer, primary_key=True) - contact_mail = sqlalchemy.Column(sqlalchemy.Text, nullable=True) - copyright = sqlalchemy.Column(sqlalchemy.Text, nullable=True) - type = sqlalchemy.Column(sqlalchemy.Text, nullable=False) - distr_type = sqlalchemy.Column(sqlalchemy.Text, nullable=False) - distr_version = sqlalchemy.Column(sqlalchemy.Text, nullable=False) - module_build_index = sqlalchemy.Column(sqlalchemy.Integer, default=1) - modularity = sqlalchemy.Column(JSONB, nullable=True) - test_dist_name = sqlalchemy.Column(sqlalchemy.Text, nullable=False) - name = sqlalchemy.Column( + id: Mapped[int] = mapped_column(sqlalchemy.Integer, primary_key=True) + contact_mail: Mapped[Optional[str]] = mapped_column( + sqlalchemy.Text, nullable=True + ) + copyright: Mapped[Optional[str]] = mapped_column( + sqlalchemy.Text, nullable=True + ) + type: Mapped[str] = mapped_column(sqlalchemy.Text, nullable=False) + distr_type: Mapped[str] = mapped_column(sqlalchemy.Text, nullable=False) + distr_version: Mapped[str] = mapped_column(sqlalchemy.Text, nullable=False) + module_build_index: Mapped[int] = mapped_column( + sqlalchemy.Integer, default=1 + ) + modularity: Mapped[Optional[Dict[str, Any]]] = mapped_column( + JSONB, nullable=True + ) + test_dist_name: Mapped[str] = mapped_column( + sqlalchemy.Text, nullable=False + ) + name: Mapped[str] = mapped_column( sqlalchemy.Text, nullable=False, unique=True, index=True ) - priority = sqlalchemy.Column(sqlalchemy.Integer, nullable=True) - arch_list = sqlalchemy.Column(JSONB, nullable=False) - copy_priority_arches = sqlalchemy.Column(JSONB, nullable=True) - weak_arch_list = sqlalchemy.Column(JSONB, nullable=True) - data = sqlalchemy.Column(JSONB, nullable=False) - is_reference = sqlalchemy.Column( + priority: Mapped[Optional[str]] = mapped_column( + sqlalchemy.Integer, nullable=True + ) + arch_list: Mapped[Dict[str, Any]] = mapped_column(JSONB, nullable=False) + copy_priority_arches: Mapped[Optional[Dict[str, Any]]] = mapped_column( + JSONB, nullable=True + ) + weak_arch_list: Mapped[Optional[Dict[str, any]]] = mapped_column( + JSONB, nullable=True + ) + data: Mapped[Dict[str, Any]] = mapped_column(JSONB, nullable=False) + is_reference: Mapped[Optional[bool]] = mapped_column( sqlalchemy.Boolean, default=False, nullable=True ) - reference_platforms = relationship( + reference_platforms: Mapped[List["Platform"]] = relationship( "Platform", secondary=ReferencePlatforms, primaryjoin=(ReferencePlatforms.c.platform_id == id), secondaryjoin=(ReferencePlatforms.c.refefence_platform_id == id), ) - repos = relationship("Repository", secondary=PlatformRepo) - sign_keys = relationship("SignKey", back_populates="platform") - roles = relationship("UserRole", secondary=PlatformRoleMapping) + repos: Mapped[List["Repository"]] = relationship( + "Repository", secondary=PlatformRepo + ) + sign_keys: Mapped[List["SignKey"]] = relationship( + "SignKey", back_populates="platform" + ) + roles: Mapped[List["UserRole"]] = relationship( + "UserRole", secondary=PlatformRoleMapping + ) class CustomRepoRepr(Base): @@ -274,31 +301,33 @@ class Repository(CustomRepoRepr, PermissionsMixin): ), ) - id = sqlalchemy.Column(sqlalchemy.Integer, primary_key=True) - name = sqlalchemy.Column(sqlalchemy.Text, nullable=False) - arch = sqlalchemy.Column(sqlalchemy.Text, nullable=False) - url = sqlalchemy.Column(sqlalchemy.Text, nullable=False) - type = sqlalchemy.Column(sqlalchemy.Text, nullable=False) - debug = sqlalchemy.Column(sqlalchemy.Boolean, default=False) - mock_enabled = sqlalchemy.Column( + id: Mapped[int] = mapped_column(sqlalchemy.Integer, primary_key=True) + name: Mapped[str] = mapped_column(sqlalchemy.Text, nullable=False) + arch: Mapped[str] = mapped_column(sqlalchemy.Text, nullable=False) + url: Mapped[str] = mapped_column(sqlalchemy.Text, nullable=False) + type: Mapped[str] = mapped_column(sqlalchemy.Text, nullable=False) + debug: Mapped[bool] = mapped_column(sqlalchemy.Boolean, default=False) + mock_enabled: Mapped[Optional[bool]] = mapped_column( sqlalchemy.Boolean, default=True, nullable=True, ) - production = sqlalchemy.Column( + production: Mapped[Optional[bool]] = mapped_column( sqlalchemy.Boolean, default=False, nullable=True ) - pulp_href = sqlalchemy.Column(sqlalchemy.Text) - export_path = sqlalchemy.Column(sqlalchemy.Text, nullable=True) - priority = sqlalchemy.Column( + pulp_href: Mapped[Optional[str]] = mapped_column(sqlalchemy.Text) + export_path: Mapped[Optional[str]] = mapped_column( + sqlalchemy.Text, nullable=True + ) + priority: Mapped[int] = mapped_column( sqlalchemy.Integer, default=10, nullable=False ) - platform_id = sqlalchemy.Column( + platform_id: Mapped[Optional[int]] = mapped_column( sqlalchemy.Integer, sqlalchemy.ForeignKey("platforms.id"), nullable=True, ) - platform = relationship("Platform") + platform: Mapped["Platform"] = relationship("Platform") class RepositoryRemote(CustomRepoRepr): @@ -312,11 +341,11 @@ class RepositoryRemote(CustomRepoRepr): ), ] - id = sqlalchemy.Column(sqlalchemy.Integer, primary_key=True) - name = sqlalchemy.Column(sqlalchemy.Text, nullable=False) - arch = sqlalchemy.Column(sqlalchemy.Text, nullable=False) - url = sqlalchemy.Column(sqlalchemy.Text, nullable=False) - pulp_href = sqlalchemy.Column(sqlalchemy.Text, nullable=False) + id: Mapped[int] = mapped_column(sqlalchemy.Integer, primary_key=True) + name: Mapped[str] = mapped_column(sqlalchemy.Text, nullable=False) + arch: Mapped[str] = mapped_column(sqlalchemy.Text, nullable=False) + url: Mapped[str] = mapped_column(sqlalchemy.Text, nullable=False) + pulp_href: Mapped[str] = mapped_column(sqlalchemy.Text, nullable=False) BuildRepo = sqlalchemy.Table( @@ -336,7 +365,6 @@ class RepositoryRemote(CustomRepoRepr): ), ) - BuildDependency = sqlalchemy.Table( "build_dependency", Base.metadata, @@ -358,28 +386,34 @@ class RepositoryRemote(CustomRepoRepr): class Build(PermissionsMixin, TeamMixin, Base): __tablename__ = "builds" - id = sqlalchemy.Column(sqlalchemy.Integer, primary_key=True) - created_at = sqlalchemy.Column( + id: Mapped[int] = mapped_column(sqlalchemy.Integer, primary_key=True) + created_at: Mapped[datetime.datetime] = mapped_column( sqlalchemy.DateTime, nullable=False, default=func.current_timestamp(), ) - finished_at = sqlalchemy.Column(sqlalchemy.DateTime, nullable=True) - tasks = relationship("BuildTask", back_populates="build") - sign_tasks = relationship( + finished_at: Mapped[Optional[datetime.datetime]] = mapped_column( + sqlalchemy.DateTime, nullable=True + ) + tasks: Mapped[List["BuildTask"]] = relationship( + "BuildTask", back_populates="build" + ) + sign_tasks: Mapped[List["SignTask"]] = relationship( "SignTask", back_populates="build", order_by="SignTask.id", ) - repos = relationship("Repository", secondary=BuildRepo) - linked_builds = relationship( + repos: Mapped[List["Repository"]] = relationship( + "Repository", secondary=BuildRepo + ) + linked_builds: Mapped[List["Build"]] = relationship( "Build", secondary=BuildDependency, primaryjoin=(BuildDependency.c.build_id == id), secondaryjoin=(BuildDependency.c.build_dependency == id), ) - mock_options = sqlalchemy.Column(JSONB) - release_id = sqlalchemy.Column( + mock_options: Mapped[Dict[str, Any]] = mapped_column(JSONB) + release_id: Mapped[Optional[int]] = mapped_column( sqlalchemy.Integer, sqlalchemy.ForeignKey( "build_releases.id", @@ -387,25 +421,29 @@ class Build(PermissionsMixin, TeamMixin, Base): ), nullable=True, ) - release = relationship("Release") - source_rpms = relationship("SourceRpm", back_populates="build") - binary_rpms = relationship("BinaryRpm", back_populates="build") - platform_flavors = relationship( + release: Mapped["Release"] = relationship("Release") + source_rpms: Mapped[List["SourceRpm"]] = relationship( + "SourceRpm", back_populates="build" + ) + binary_rpms: Mapped[List["BinaryRpm"]] = relationship( + "BinaryRpm", back_populates="build" + ) + platform_flavors: Mapped[List["PlatformFlavour"]] = relationship( "PlatformFlavour", secondary=BuildPlatformFlavour, ) - products = relationship( + products: Mapped[List["Product"]] = relationship( "Product", secondary="product_packages", back_populates="builds", cascade="all, delete", passive_deletes=True, ) - released = sqlalchemy.Column(sqlalchemy.Boolean, default=False) - signed = sqlalchemy.Column( + released: Mapped[bool] = mapped_column(sqlalchemy.Boolean, default=False) + signed: Mapped[Optional[bool]] = mapped_column( sqlalchemy.Boolean, default=False, nullable=True ) - cancel_testing = sqlalchemy.Column( + cancel_testing: Mapped[bool] = mapped_column( sqlalchemy.Boolean, default=False, nullable=False ) @@ -431,13 +469,13 @@ class Build(PermissionsMixin, TeamMixin, Base): class BuildTask(TimeMixin, Base): __tablename__ = "build_tasks" - id = sqlalchemy.Column(sqlalchemy.Integer, primary_key=True) - ts = sqlalchemy.Column( + id: Mapped[int] = mapped_column(sqlalchemy.Integer, primary_key=True) + ts: Mapped[Optional[datetime.datetime]] = mapped_column( sqlalchemy.DateTime, nullable=True, index=True, ) - build_id = sqlalchemy.Column( + build_id: Mapped[int] = mapped_column( sqlalchemy.Integer, sqlalchemy.ForeignKey("builds.id"), # saw https://stackoverflow.com/questions/ @@ -445,85 +483,99 @@ class BuildTask(TimeMixin, Base): nullable=False, index=True, ) - platform_id = sqlalchemy.Column( + platform_id: Mapped[int] = mapped_column( sqlalchemy.Integer, sqlalchemy.ForeignKey("platforms.id"), nullable=False, ) - ref_id = sqlalchemy.Column( + ref_id: Mapped[int] = mapped_column( sqlalchemy.Integer, sqlalchemy.ForeignKey("build_task_refs.id"), nullable=False, index=True, ) - rpm_module_id = sqlalchemy.Column( + rpm_module_id: Mapped[int] = mapped_column( sqlalchemy.Integer, sqlalchemy.ForeignKey("rpm_module.id"), nullable=True, ) - status = sqlalchemy.Column( + status: Mapped[int] = mapped_column( sqlalchemy.Integer, nullable=False, index=True, ) - index = sqlalchemy.Column(sqlalchemy.Integer, nullable=False) - arch = sqlalchemy.Column( + index: Mapped[int] = mapped_column(sqlalchemy.Integer, nullable=False) + arch: Mapped[str] = mapped_column( sqlalchemy.VARCHAR(length=50), nullable=False, index=True, ) - is_secure_boot = sqlalchemy.Column( + is_secure_boot: Mapped[Optional[str]] = mapped_column( sqlalchemy.Boolean, default=False, nullable=True ) - mock_options = sqlalchemy.Column(JSONB) - ref = relationship("BuildTaskRef") - alma_commit_cas_hash = sqlalchemy.Column(sqlalchemy.Text, nullable=True) - is_cas_authenticated = sqlalchemy.Column( + mock_options: Mapped[Dict[str, Any]] = mapped_column(JSONB) + ref: Mapped["BuildTaskRef"] = relationship("BuildTaskRef") + alma_commit_cas_hash: Mapped[Optional[str]] = mapped_column( + sqlalchemy.Text, nullable=True + ) + is_cas_authenticated: Mapped[Optional[bool]] = mapped_column( sqlalchemy.Boolean, default=False, nullable=True ) - artifacts = relationship("BuildTaskArtifact", back_populates="build_task") - platform = relationship("Platform") - build = relationship("Build", back_populates="tasks") - dependencies = relationship( + artifacts: Mapped[List["BuildTaskArtifact"]] = relationship( + "BuildTaskArtifact", back_populates="build_task" + ) + platform: Mapped["Platform"] = relationship("Platform") + build: Mapped["Build"] = relationship("Build", back_populates="tasks") + dependencies: Mapped[List["BuildTask"]] = relationship( "BuildTask", secondary=BuildTaskDependency, primaryjoin=(BuildTaskDependency.c.build_task_id == id), secondaryjoin=(BuildTaskDependency.c.build_task_dependency == id), ) - test_tasks = relationship( + test_tasks: Mapped[List["TestTask"]] = relationship( "TestTask", back_populates="build_task", order_by="TestTask.revision" ) - rpm_module = relationship("RpmModule") - performance_stats: "PerformanceStats" = relationship( + rpm_module: Mapped["RpmModule"] = relationship("RpmModule") + performance_stats: Mapped[List["PerformanceStats"]] = relationship( "PerformanceStats", back_populates="build_task", ) - built_srpm_url = sqlalchemy.Column(sqlalchemy.VARCHAR, nullable=True) - error = sqlalchemy.Column(sqlalchemy.Text, nullable=True, default=None) + built_srpm_url: Mapped[Optional[str]] = mapped_column( + sqlalchemy.VARCHAR, nullable=True + ) + error: Mapped[Optional[str]] = mapped_column( + sqlalchemy.Text, nullable=True, default=None + ) class BuildTaskRef(Base): __tablename__ = "build_task_refs" - id = sqlalchemy.Column(sqlalchemy.Integer, primary_key=True) - url = sqlalchemy.Column(sqlalchemy.TEXT, nullable=False) - git_ref = sqlalchemy.Column(sqlalchemy.TEXT) - ref_type = sqlalchemy.Column(sqlalchemy.Integer) - git_commit_hash = sqlalchemy.Column(sqlalchemy.TEXT, nullable=True) - test_configuration = sqlalchemy.Column(JSONB, nullable=True) + id: Mapped[int] = mapped_column(sqlalchemy.Integer, primary_key=True) + url: Mapped[str] = mapped_column(sqlalchemy.TEXT, nullable=False) + git_ref: Mapped[Optional[str]] = mapped_column(sqlalchemy.TEXT) + ref_type: Mapped[Optional[int]] = mapped_column(sqlalchemy.Integer) + git_commit_hash: Mapped[Optional[str]] = mapped_column( + sqlalchemy.TEXT, nullable=True + ) + test_configuration: Mapped[Optional[Dict[str, Any]]] = mapped_column( + JSONB, nullable=True + ) class RpmModule(Base): __tablename__ = "rpm_module" - id = sqlalchemy.Column(sqlalchemy.Integer, primary_key=True) - name = sqlalchemy.Column(sqlalchemy.TEXT, nullable=False) - version = sqlalchemy.Column(sqlalchemy.TEXT, nullable=False) - stream = sqlalchemy.Column(sqlalchemy.TEXT, nullable=False) - context = sqlalchemy.Column(sqlalchemy.TEXT, nullable=False) - arch = sqlalchemy.Column(sqlalchemy.TEXT, nullable=False) - pulp_href = sqlalchemy.Column(sqlalchemy.TEXT, nullable=False) - sha256 = sqlalchemy.Column(sqlalchemy.VARCHAR(64), nullable=True) + id: Mapped[int] = mapped_column(sqlalchemy.Integer, primary_key=True) + name: Mapped[str] = mapped_column(sqlalchemy.TEXT, nullable=False) + version: Mapped[str] = mapped_column(sqlalchemy.TEXT, nullable=False) + stream: Mapped[str] = mapped_column(sqlalchemy.TEXT, nullable=False) + context: Mapped[str] = mapped_column(sqlalchemy.TEXT, nullable=False) + arch: Mapped[str] = mapped_column(sqlalchemy.TEXT, nullable=False) + pulp_href: Mapped[str] = mapped_column(sqlalchemy.TEXT, nullable=False) + sha256: Mapped[Optional[str]] = mapped_column( + sqlalchemy.VARCHAR(64), nullable=True + ) @property def nvsca(self): @@ -536,19 +588,23 @@ def nvsca(self): class BuildTaskArtifact(Base): __tablename__ = "build_artifacts" - id = sqlalchemy.Column(sqlalchemy.Integer, primary_key=True) - build_task_id = sqlalchemy.Column( + id: Mapped[int] = mapped_column(sqlalchemy.Integer, primary_key=True) + build_task_id: Mapped[int] = mapped_column( sqlalchemy.Integer, sqlalchemy.ForeignKey("build_tasks.id"), nullable=False, index=True, ) - name = sqlalchemy.Column(sqlalchemy.Text, nullable=False) - type = sqlalchemy.Column(sqlalchemy.Text, nullable=False) - href = sqlalchemy.Column(sqlalchemy.Text, nullable=False) - build_task = relationship("BuildTask", back_populates="artifacts") - cas_hash = sqlalchemy.Column(sqlalchemy.Text, nullable=True) - sign_key_id = sqlalchemy.Column( + name: Mapped[str] = mapped_column(sqlalchemy.Text, nullable=False) + type: Mapped[str] = mapped_column(sqlalchemy.Text, nullable=False) + href: Mapped[str] = mapped_column(sqlalchemy.Text, nullable=False) + build_task: Mapped["BuildTask"] = relationship( + "BuildTask", back_populates="artifacts" + ) + cas_hash: Mapped[Optional[str]] = mapped_column( + sqlalchemy.Text, nullable=True + ) + sign_key_id: Mapped[Optional[int]] = mapped_column( sqlalchemy.Integer, sqlalchemy.ForeignKey( "sign_keys.id", @@ -556,7 +612,9 @@ class BuildTaskArtifact(Base): ), nullable=True, ) - sign_key = relationship("SignKey", back_populates="build_task_artifacts") + sign_key: Mapped["SignKey"] = relationship( + "SignKey", back_populates="build_task_artifacts" + ) def name_as_dict(self) -> dict: result = re.search( @@ -574,51 +632,61 @@ def name_as_dict(self) -> dict: class SourceRpm(Base): __tablename__ = "source_rpms" - id = sqlalchemy.Column(sqlalchemy.Integer, primary_key=True) - build_id = sqlalchemy.Column( + id: Mapped[int] = mapped_column(sqlalchemy.Integer, primary_key=True) + build_id: Mapped[int] = mapped_column( sqlalchemy.Integer, sqlalchemy.ForeignKey("builds.id"), nullable=False, index=True, ) - build = relationship("Build", back_populates="source_rpms") - artifact_id = sqlalchemy.Column( + build: Mapped["Build"] = relationship( + "Build", back_populates="source_rpms" + ) + artifact_id: Mapped[int] = mapped_column( sqlalchemy.Integer, sqlalchemy.ForeignKey("build_artifacts.id"), nullable=False, ) - artifact = relationship("BuildTaskArtifact") - binary_rpms = relationship("BinaryRpm", back_populates="source_rpm") + artifact: Mapped["BuildTaskArtifact"] = relationship("BuildTaskArtifact") + binary_rpms: Mapped[List["BinaryRpm"]] = relationship( + "BinaryRpm", back_populates="source_rpm" + ) class BinaryRpm(Base): __tablename__ = "binary_rpms" - id = sqlalchemy.Column(sqlalchemy.Integer, primary_key=True) - build_id = sqlalchemy.Column( + id: Mapped[int] = mapped_column(sqlalchemy.Integer, primary_key=True) + build_id: Mapped[int] = mapped_column( sqlalchemy.Integer, sqlalchemy.ForeignKey("builds.id"), nullable=False ) - build = relationship("Build", back_populates="binary_rpms") - artifact_id = sqlalchemy.Column( + build: Mapped["Build"] = relationship( + "Build", back_populates="binary_rpms" + ) + artifact_id: Mapped[int] = mapped_column( sqlalchemy.Integer, sqlalchemy.ForeignKey("build_artifacts.id"), nullable=False, ) - artifact = relationship("BuildTaskArtifact") - source_rpm_id = sqlalchemy.Column( + artifact: Mapped["BuildTaskArtifact"] = relationship("BuildTaskArtifact") + source_rpm_id: Mapped[int] = mapped_column( sqlalchemy.Integer, sqlalchemy.ForeignKey("source_rpms.id"), nullable=False, ) - source_rpm = relationship("SourceRpm", back_populates="binary_rpms") + source_rpm: Mapped["SourceRpm"] = relationship( + "SourceRpm", back_populates="binary_rpms" + ) class UserAction(Base): __tablename__ = "user_actions" - id = sqlalchemy.Column(sqlalchemy.Integer, primary_key=True) - name = sqlalchemy.Column(sqlalchemy.String(100), unique=True) - description = sqlalchemy.Column(sqlalchemy.TEXT, nullable=True) + id: Mapped[int] = mapped_column(sqlalchemy.Integer, primary_key=True) + name: Mapped[str] = mapped_column(sqlalchemy.String(100), unique=True) + description: Mapped[Optional[str]] = mapped_column( + sqlalchemy.TEXT, nullable=True + ) ActionRoleMapping = sqlalchemy.Table( @@ -650,9 +718,11 @@ class UserAction(Base): class UserRole(Base): __tablename__ = "user_roles" - id = sqlalchemy.Column(sqlalchemy.Integer, primary_key=True) - name = sqlalchemy.Column(sqlalchemy.String(100), unique=True) - actions = relationship("UserAction", secondary=ActionRoleMapping) + id: Mapped[int] = mapped_column(sqlalchemy.Integer, primary_key=True) + name: Mapped[str] = mapped_column(sqlalchemy.String(100), unique=True) + actions: Mapped[List["UserAction"]] = relationship( + "UserAction", secondary=ActionRoleMapping + ) def __repr__(self): return f"{self.__class__.__name__}: {self.id} {self.name}" @@ -754,16 +824,16 @@ def __repr__(self): class UserOauthAccount(SQLAlchemyBaseOAuthAccountTable[int], Base): __tablename__ = "user_oauth_accounts" - id = sqlalchemy.Column(sqlalchemy.Integer, primary_key=True) + id: Mapped[int] = mapped_column(sqlalchemy.Integer, primary_key=True) # Override SQLAlchemyBaseOAuthAccountTable access_token column length - access_token = sqlalchemy.Column( + access_token: Mapped[str] = mapped_column( sqlalchemy.VARCHAR(length=2048), nullable=False, ) @declared_attr - def user_id(cls): - return sqlalchemy.Column( + def user_id(cls) -> Mapped[int]: + return mapped_column( sqlalchemy.Integer, sqlalchemy.ForeignKey("users.id", ondelete="cascade"), nullable=False, @@ -773,13 +843,13 @@ def user_id(cls): class UserAccessToken(SQLAlchemyBaseAccessTokenTable[int], Base): __tablename__ = "user_access_tokens" - id = sqlalchemy.Column( + id: Mapped[int] = mapped_column( sqlalchemy.Integer, primary_key=True, autoincrement=True ) @declared_attr - def user_id(cls): - return sqlalchemy.Column( + def user_id(cls) -> Mapped[int]: + return mapped_column( sqlalchemy.Integer, sqlalchemy.ForeignKey("users.id", ondelete="cascade"), nullable=False, @@ -789,38 +859,52 @@ def user_id(cls): class User(SQLAlchemyBaseUserTable[int], Base): __tablename__ = "users" - id = sqlalchemy.Column(sqlalchemy.Integer, primary_key=True) - username = sqlalchemy.Column(sqlalchemy.TEXT, nullable=True) - first_name = sqlalchemy.Column(sqlalchemy.String(320), nullable=True) - last_name = sqlalchemy.Column(sqlalchemy.String(320), nullable=True) + id: Mapped[int] = mapped_column(sqlalchemy.Integer, primary_key=True) + username: Mapped[Optional[str]] = mapped_column( + sqlalchemy.TEXT, nullable=True + ) + first_name: Mapped[Optional[str]] = mapped_column( + sqlalchemy.String(320), nullable=True + ) + last_name: Mapped[Optional[str]] = mapped_column( + sqlalchemy.String(320), nullable=True + ) # Override SQLAlchemyBaseUserTable email attribute to keep current type - email = sqlalchemy.Column( + email: Mapped[str] = mapped_column( sqlalchemy.TEXT, unique=True, index=True, nullable=False, ) - hashed_password: str = sqlalchemy.Column( + hashed_password: Mapped[Optional[str]] = mapped_column( sqlalchemy.String(length=1024), nullable=True, ) - roles = relationship("UserRole", secondary=UserRoleMapping) - teams = relationship( + roles: Mapped[List["UserRole"]] = relationship( + "UserRole", secondary=UserRoleMapping + ) + teams: Mapped[List["Team"]] = relationship( "Team", secondary=TeamUserMapping, back_populates="members" ) - oauth_accounts = relationship("UserOauthAccount", lazy="joined") + oauth_accounts: Mapped[List["UserOauthAccount"]] = relationship( + "UserOauthAccount", lazy="joined" + ) class Team(PermissionsMixin, Base): __tablename__ = "teams" - id = sqlalchemy.Column(sqlalchemy.Integer, primary_key=True) - name = sqlalchemy.Column(sqlalchemy.Text, nullable=False, unique=True) - members = relationship( + id: Mapped[int] = mapped_column(sqlalchemy.Integer, primary_key=True) + name: Mapped[str] = mapped_column( + sqlalchemy.Text, nullable=False, unique=True + ) + members: Mapped[List["User"]] = relationship( "User", secondary=TeamUserMapping, back_populates="teams" ) - products = relationship("Product", back_populates="team") - roles = relationship( + products: Mapped[List["Product"]] = relationship( + "Product", back_populates="team" + ) + roles: Mapped[List["UserRole"]] = relationship( "UserRole", secondary=TeamRoleMapping, cascade="all, delete", @@ -850,7 +934,6 @@ class Team(PermissionsMixin, Base): ), ) - ProductBuilds = sqlalchemy.Table( "product_packages", Base.metadata, @@ -874,7 +957,6 @@ class Team(PermissionsMixin, Base): ), ) - ProductPlatforms = sqlalchemy.Table( "product_platforms", Base.metadata, @@ -902,31 +984,39 @@ class Team(PermissionsMixin, Base): class Product(PermissionsMixin, TeamMixin, Base): __tablename__ = "products" - id = sqlalchemy.Column(sqlalchemy.Integer, primary_key=True) - name = sqlalchemy.Column(sqlalchemy.Text, nullable=False, unique=True) + id: Mapped[int] = mapped_column(sqlalchemy.Integer, primary_key=True) + name: Mapped[str] = mapped_column( + sqlalchemy.Text, nullable=False, unique=True + ) # FIXME: change nullable to False after population - title = sqlalchemy.Column(sqlalchemy.String(100), nullable=True) - description = sqlalchemy.Column(sqlalchemy.Text, nullable=True) - team = relationship("Team", back_populates="products") - is_community = sqlalchemy.Column( + title: Mapped[Optional[str]] = mapped_column( + sqlalchemy.String(100), nullable=True + ) + description: Mapped[Optional[str]] = mapped_column( + sqlalchemy.Text, nullable=True + ) + team: Mapped["Team"] = relationship("Team", back_populates="products") + is_community: Mapped[bool] = mapped_column( sqlalchemy.Boolean, nullable=False, default=True ) - roles = relationship("UserRole", secondary=ProductRoleMapping) - repositories = relationship( + roles: Mapped[List["UserRole"]] = relationship( + "UserRole", secondary=ProductRoleMapping + ) + repositories: Mapped[List["Repository"]] = relationship( "Repository", secondary=ProductRepositories, cascade="all, delete", ) - platforms = relationship( + platforms: Mapped[List["Platform"]] = relationship( "Platform", secondary=ProductPlatforms, ) - builds = relationship( + builds: Mapped[List["Build"]] = relationship( "Build", secondary=ProductBuilds, back_populates="products", ) - sign_keys = relationship( + sign_keys: Mapped[List["SignKey"]] = relationship( "SignKey", back_populates="product", cascade="all, delete-orphan", @@ -944,38 +1034,50 @@ def pulp_base_distro_name(self) -> str: class TestTask(TimeMixin, Base): __tablename__ = "test_tasks" - id = sqlalchemy.Column(sqlalchemy.Integer, primary_key=True) - package_name = sqlalchemy.Column(sqlalchemy.TEXT, nullable=False) - package_version = sqlalchemy.Column(sqlalchemy.TEXT, nullable=False) - package_release = sqlalchemy.Column(sqlalchemy.TEXT, nullable=True) - env_arch = sqlalchemy.Column(sqlalchemy.TEXT, nullable=False) - build_task_id = sqlalchemy.Column( + id: Mapped[int] = mapped_column(sqlalchemy.Integer, primary_key=True) + package_name: Mapped[str] = mapped_column(sqlalchemy.TEXT, nullable=False) + package_version: Mapped[str] = mapped_column( + sqlalchemy.TEXT, nullable=False + ) + package_release: Mapped[Optional[str]] = mapped_column( + sqlalchemy.TEXT, nullable=True + ) + env_arch: Mapped[str] = mapped_column(sqlalchemy.TEXT, nullable=False) + build_task_id: Mapped[int] = mapped_column( sqlalchemy.Integer, sqlalchemy.ForeignKey("build_tasks.id"), nullable=False, index=True, ) - build_task = relationship("BuildTask", back_populates="test_tasks") - status = sqlalchemy.Column( + build_task: Mapped["BuildTask"] = relationship( + "BuildTask", back_populates="test_tasks" + ) + status: Mapped[int] = mapped_column( sqlalchemy.Integer, nullable=False, index=True, ) - alts_response = sqlalchemy.Column(JSONB, nullable=True) - revision = sqlalchemy.Column( + alts_response: Mapped[Optional[Dict[str, Any]]] = mapped_column( + JSONB, nullable=True + ) + revision: Mapped[int] = mapped_column( sqlalchemy.Integer, nullable=False, index=True, ) - artifacts = relationship("TestTaskArtifact", back_populates="test_task") - repository_id = sqlalchemy.Column( + artifacts: Mapped[List["TestTaskArtifact"]] = relationship( + "TestTaskArtifact", back_populates="test_task" + ) + repository_id: Mapped[int] = mapped_column( sqlalchemy.Integer, sqlalchemy.ForeignKey("repositories.id", name="test_task_repo_fk"), nullable=True, ) - repository = relationship("Repository") - scheduled_at = sqlalchemy.Column(sqlalchemy.DateTime, nullable=True) - performance_stats: "PerformanceStats" = relationship( + repository: Mapped["Repository"] = relationship("Repository") + scheduled_at: Mapped[Optional[datetime.datetime]] = mapped_column( + sqlalchemy.DateTime, nullable=True + ) + performance_stats: Mapped[List["PerformanceStats"]] = relationship( "PerformanceStats", back_populates="test_task", ) @@ -983,15 +1085,17 @@ class TestTask(TimeMixin, Base): class TestTaskArtifact(Base): __tablename__ = "test_task_artifacts" - id = sqlalchemy.Column(sqlalchemy.Integer, primary_key=True) - test_task_id = sqlalchemy.Column( + id: Mapped[int] = mapped_column(sqlalchemy.Integer, primary_key=True) + test_task_id: Mapped[int] = mapped_column( sqlalchemy.Integer, sqlalchemy.ForeignKey("test_tasks.id"), nullable=False, ) - test_task = relationship("TestTask", back_populates="artifacts") - name = sqlalchemy.Column(sqlalchemy.Text, nullable=False) - href = sqlalchemy.Column(sqlalchemy.Text, nullable=False) + test_task: Mapped["TestTask"] = relationship( + "TestTask", back_populates="artifacts" + ) + name: Mapped[str] = mapped_column(sqlalchemy.Text, nullable=False) + href: Mapped[str] = mapped_column(sqlalchemy.Text, nullable=False) class PackageTestRepository(Base): @@ -1003,11 +1107,11 @@ class PackageTestRepository(Base): name="package_test_repo_uix", ), ] - id = sqlalchemy.Column(sqlalchemy.Integer, primary_key=True) - package_name = sqlalchemy.Column(sqlalchemy.Text, nullable=False) - folder_name = sqlalchemy.Column(sqlalchemy.Text, nullable=False) - url = sqlalchemy.Column(sqlalchemy.Text, nullable=False) - test_repository_id = sqlalchemy.Column( + id: Mapped[int] = mapped_column(sqlalchemy.Integer, primary_key=True) + package_name: Mapped[str] = mapped_column(sqlalchemy.Text, nullable=False) + folder_name: Mapped[str] = mapped_column(sqlalchemy.Text, nullable=False) + url: Mapped[str] = mapped_column(sqlalchemy.Text, nullable=False) + test_repository_id: Mapped[int] = mapped_column( sqlalchemy.Integer, sqlalchemy.ForeignKey( "test_repositories.id", @@ -1016,17 +1120,25 @@ class PackageTestRepository(Base): ), nullable=False, ) - test_repository = relationship("TestRepository", back_populates="packages") + test_repository: Mapped["TestRepository"] = relationship( + "TestRepository", back_populates="packages" + ) class TestRepository(Base): __tablename__ = "test_repositories" - id = sqlalchemy.Column(sqlalchemy.Integer, primary_key=True) - name = sqlalchemy.Column(sqlalchemy.Text, nullable=False, unique=True) - url = sqlalchemy.Column(sqlalchemy.Text, nullable=False, unique=True) - tests_dir = sqlalchemy.Column(sqlalchemy.Text, nullable=False) - tests_prefix = sqlalchemy.Column(sqlalchemy.Text, nullable=True) - packages: List["PackageTestRepository"] = relationship( + id: Mapped[int] = mapped_column(sqlalchemy.Integer, primary_key=True) + name: Mapped[str] = mapped_column( + sqlalchemy.Text, nullable=False, unique=True + ) + url: Mapped[str] = mapped_column( + sqlalchemy.Text, nullable=False, unique=True + ) + tests_dir: Mapped[str] = mapped_column(sqlalchemy.Text, nullable=False) + tests_prefix: Mapped[Optional[str]] = mapped_column( + sqlalchemy.Text, nullable=True + ) + packages: Mapped[List["PackageTestRepository"]] = relationship( "PackageTestRepository", back_populates="test_repository", cascade="all, delete", @@ -1036,29 +1148,29 @@ class TestRepository(Base): class Release(PermissionsMixin, TeamMixin, TimeMixin, Base): __tablename__ = "build_releases" - id = sqlalchemy.Column(sqlalchemy.Integer, primary_key=True) - build_ids = sqlalchemy.Column( + id: Mapped[int] = mapped_column(sqlalchemy.Integer, primary_key=True) + build_ids: Mapped[List[int]] = mapped_column( sqlalchemy.ARRAY(sqlalchemy.Integer, dimensions=1), nullable=False ) - created_at = sqlalchemy.Column( + created_at: Mapped[Optional[datetime.datetime]] = mapped_column( sqlalchemy.DateTime, nullable=True, default=func.current_timestamp(), ) - build_task_ids = sqlalchemy.Column( + build_task_ids: Mapped[List[int]] = mapped_column( sqlalchemy.ARRAY(sqlalchemy.Integer, dimensions=1), nullable=True ) - reference_platform_id = sqlalchemy.Column( + reference_platform_id: Mapped[Optional[int]] = mapped_column( sqlalchemy.Integer, nullable=True, ) - platform_id = sqlalchemy.Column( + platform_id: Mapped[int] = mapped_column( sqlalchemy.Integer, sqlalchemy.ForeignKey("platforms.id"), nullable=False, ) - platform = relationship("Platform") - product_id = sqlalchemy.Column( + platform: Mapped["Platform"] = relationship("Platform") + product_id: Mapped[int] = mapped_column( sqlalchemy.Integer, sqlalchemy.ForeignKey( "products.id", @@ -1066,12 +1178,14 @@ class Release(PermissionsMixin, TeamMixin, TimeMixin, Base): ), nullable=False, ) - product = relationship("Product") - plan = sqlalchemy.Column(JSONB, nullable=True) - status = sqlalchemy.Column( + product: Mapped["Product"] = relationship("Product") + plan: Mapped[Optional[Dict[str, Any]]] = mapped_column( + JSONB, nullable=True + ) + status: Mapped[int] = mapped_column( sqlalchemy.Integer, default=ReleaseStatus.SCHEDULED ) - performance_stats: List["PerformanceStats"] = relationship( + performance_stats: Mapped[List["PerformanceStats"]] = relationship( "PerformanceStats", back_populates="release", ) @@ -1106,22 +1220,26 @@ class Release(PermissionsMixin, TeamMixin, TimeMixin, Base): class SignKey(PermissionsMixin, Base): __tablename__ = "sign_keys" - id = sqlalchemy.Column(sqlalchemy.Integer, primary_key=True) - name = sqlalchemy.Column(sqlalchemy.Text) + id: Mapped[int] = mapped_column(sqlalchemy.Integer, primary_key=True) + name: Mapped[str] = mapped_column(sqlalchemy.Text) # FIXME: change nullable to False after population - is_community = sqlalchemy.Column( + is_community: Mapped[Optional[bool]] = mapped_column( sqlalchemy.Boolean, nullable=True, default=False, ) - description = sqlalchemy.Column(sqlalchemy.Text, nullable=True) - keyid = sqlalchemy.Column(sqlalchemy.String(16), unique=True) - fingerprint = sqlalchemy.Column(sqlalchemy.String(40), unique=True) - public_url = sqlalchemy.Column(sqlalchemy.Text) - inserted = sqlalchemy.Column( + description: Mapped[Optional[str]] = mapped_column( + sqlalchemy.Text, nullable=True + ) + keyid: Mapped[str] = mapped_column(sqlalchemy.String(16), unique=True) + fingerprint: Mapped[str] = mapped_column( + sqlalchemy.String(40), unique=True + ) + public_url: Mapped[str] = mapped_column(sqlalchemy.Text) + inserted: Mapped[datetime.datetime] = mapped_column( sqlalchemy.DateTime, default=datetime.datetime.utcnow() ) - product_id = sqlalchemy.Column( + product_id: Mapped[Optional[int]] = mapped_column( sqlalchemy.Integer, sqlalchemy.ForeignKey( 'products.id', @@ -1129,8 +1247,10 @@ class SignKey(PermissionsMixin, Base): ), nullable=True, ) - product = relationship('Product', back_populates='sign_keys') - platform_id = sqlalchemy.Column( + product: Mapped["Product"] = relationship( + 'Product', back_populates='sign_keys' + ) + platform_id: Mapped[Optional[int]] = mapped_column( sqlalchemy.Integer, sqlalchemy.ForeignKey( "platforms.id", @@ -1138,22 +1258,30 @@ class SignKey(PermissionsMixin, Base): ), nullable=True, ) - platform = relationship("Platform", back_populates="sign_keys") - build_task_artifacts = relationship( + platform: Mapped["Platform"] = relationship( + "Platform", back_populates="sign_keys" + ) + build_task_artifacts: Mapped[List["BuildTaskArtifact"]] = relationship( "BuildTaskArtifact", back_populates="sign_key", ) - roles = relationship("UserRole", secondary=SignKeyRoleMapping) + roles: Mapped[List["UserRole"]] = relationship( + "UserRole", secondary=SignKeyRoleMapping + ) class GenKeyTask(Base): __tablename__ = "gen_key_tasks" - id = sqlalchemy.Column(sqlalchemy.Integer, primary_key=True) - status = sqlalchemy.Column(sqlalchemy.Integer, default=GenKeyStatus.IDLE) - error_message = sqlalchemy.Column(sqlalchemy.Text, nullable=True) - product = relationship("Product") - product_id = sqlalchemy.Column( + id: Mapped[int] = mapped_column(sqlalchemy.Integer, primary_key=True) + status: Mapped[int] = mapped_column( + sqlalchemy.Integer, default=GenKeyStatus.IDLE + ) + error_message: Mapped[Optional[str]] = mapped_column( + sqlalchemy.Text, nullable=True + ) + product: Mapped["Product"] = relationship("Product") + product_id: Mapped[int] = mapped_column( sqlalchemy.Integer, sqlalchemy.ForeignKey("products.id"), nullable=False, @@ -1163,66 +1291,88 @@ class GenKeyTask(Base): class SignTask(TimeMixin, Base): __tablename__ = "sign_tasks" - id = sqlalchemy.Column(sqlalchemy.Integer, primary_key=True) - build_id = sqlalchemy.Column( + id: Mapped[int] = mapped_column(sqlalchemy.Integer, primary_key=True) + build_id: Mapped[int] = mapped_column( sqlalchemy.Integer, sqlalchemy.ForeignKey("builds.id"), nullable=False ) - build = relationship("Build") - sign_key_id = sqlalchemy.Column( + build: Mapped["Build"] = relationship("Build") + sign_key_id: Mapped[int] = mapped_column( sqlalchemy.Integer, sqlalchemy.ForeignKey("sign_keys.id"), nullable=False, ) - sign_key = relationship("SignKey") - status = sqlalchemy.Column(sqlalchemy.Integer, default=SignStatus.IDLE) - ts = sqlalchemy.Column(sqlalchemy.DateTime, nullable=True) - error_message = sqlalchemy.Column(sqlalchemy.Text, nullable=True) - log_href = sqlalchemy.Column(sqlalchemy.Text, nullable=True) - stats = sqlalchemy.Column(JSONB, nullable=True) + sign_key: Mapped["SignKey"] = relationship("SignKey") + status: Mapped[int] = mapped_column( + sqlalchemy.Integer, default=SignStatus.IDLE + ) + ts: Mapped[Optional[datetime.datetime]] = mapped_column( + sqlalchemy.DateTime, nullable=True + ) + error_message: Mapped[Optional[str]] = mapped_column( + sqlalchemy.Text, nullable=True + ) + log_href: Mapped[Optional[str]] = mapped_column( + sqlalchemy.Text, nullable=True + ) + stats: Mapped[Optional[Dict[str, Any]]] = mapped_column( + JSONB, nullable=True + ) class ExportTask(Base): __tablename__ = "export_tasks" - id = sqlalchemy.Column(sqlalchemy.Integer, primary_key=True) - name = sqlalchemy.Column(sqlalchemy.Text, nullable=False) - status = sqlalchemy.Column(sqlalchemy.Integer, nullable=False) - exported_at = sqlalchemy.Column(sqlalchemy.DateTime, nullable=True) + id: Mapped[int] = mapped_column(sqlalchemy.Integer, primary_key=True) + name: Mapped[str] = mapped_column(sqlalchemy.Text, nullable=False) + status: Mapped[int] = mapped_column(sqlalchemy.Integer, nullable=False) + exported_at: Mapped[Optional[datetime.datetime]] = mapped_column( + sqlalchemy.DateTime, nullable=True + ) class RepoExporter(Base): __tablename__ = "repo_exporters" - id = sqlalchemy.Column(sqlalchemy.Integer, primary_key=True) - path = sqlalchemy.Column(sqlalchemy.Text, nullable=False) - exported_id = sqlalchemy.Column( + id: Mapped[int] = mapped_column(sqlalchemy.Integer, primary_key=True) + path: Mapped[str] = mapped_column(sqlalchemy.Text, nullable=False) + exported_id: Mapped[int] = mapped_column( sqlalchemy.Integer, sqlalchemy.ForeignKey("export_tasks.id"), nullable=False, ) - repository_id = sqlalchemy.Column( + repository_id: Mapped[int] = mapped_column( sqlalchemy.Integer, sqlalchemy.ForeignKey("repositories.id"), nullable=False, ) - repository = relationship("Repository") - fs_exporter_href = sqlalchemy.Column(sqlalchemy.Text, nullable=False) + repository: Mapped["Repository"] = relationship("Repository") + fs_exporter_href: Mapped[str] = mapped_column( + sqlalchemy.Text, nullable=False + ) class PlatformFlavour(PermissionsMixin, Base): __tablename__ = "platform_flavours" - id = sqlalchemy.Column(sqlalchemy.Integer, primary_key=True) - name = sqlalchemy.Column(sqlalchemy.Text, nullable=False, unique=True) - modularity = sqlalchemy.Column(JSONB, nullable=True) - repos = relationship("Repository", secondary=FlavourRepo) - data = sqlalchemy.Column(JSONB, nullable=True) + id: Mapped[int] = mapped_column(sqlalchemy.Integer, primary_key=True) + name: Mapped[str] = mapped_column( + sqlalchemy.Text, nullable=False, unique=True + ) + modularity: Mapped[Optional[Dict[str, Any]]] = mapped_column( + JSONB, nullable=True + ) + repos: Mapped[List["Repository"]] = relationship( + "Repository", secondary=FlavourRepo + ) + data: Mapped[Optional[Dict[str, Any]]] = mapped_column( + JSONB, nullable=True + ) class NewErrataRecord(Base): __tablename__ = "new_errata_records" - id = sqlalchemy.Column(sqlalchemy.Text, primary_key=True) - platform_id = sqlalchemy.Column( + id: Mapped[str] = mapped_column(sqlalchemy.Text, primary_key=True) + platform_id: Mapped[int] = mapped_column( sqlalchemy.Integer, sqlalchemy.ForeignKey( "platforms.id", @@ -1231,56 +1381,117 @@ class NewErrataRecord(Base): nullable=False, primary_key=True, ) - platform = relationship("Platform") - module = sqlalchemy.Column(sqlalchemy.Text, nullable=True) - release_status = sqlalchemy.Column( + platform: Mapped["Platform"] = relationship("Platform") + module: Mapped[Optional[str]] = mapped_column( + sqlalchemy.Text, nullable=True + ) + release_status: Mapped[ + Optional[ + Literal[ + ErrataReleaseStatus.NOT_RELEASED, + ErrataReleaseStatus.IN_PROGRESS, + ErrataReleaseStatus.RELEASED, + ErrataReleaseStatus.FAILED, + ] + ] + ] = mapped_column( sqlalchemy.Enum(ErrataReleaseStatus, name='erratareleasestatus'), nullable=False, ) - last_release_log = sqlalchemy.Column(sqlalchemy.Text, nullable=True) - summary = sqlalchemy.Column(sqlalchemy.Text, nullable=True) - solution = sqlalchemy.Column(sqlalchemy.Text, nullable=True) - - freezed = sqlalchemy.Column(sqlalchemy.Boolean, nullable=True) - - issued_date = sqlalchemy.Column(sqlalchemy.DateTime, nullable=False) - updated_date = sqlalchemy.Column(sqlalchemy.DateTime, nullable=False) - description = sqlalchemy.Column(sqlalchemy.Text, nullable=True) - original_description = sqlalchemy.Column(sqlalchemy.Text, nullable=False) - title = sqlalchemy.Column(sqlalchemy.Text, nullable=True) - oval_title = sqlalchemy.Column(sqlalchemy.Text, nullable=True) - original_title = sqlalchemy.Column(sqlalchemy.Text, nullable=False) - contact_mail = sqlalchemy.Column(sqlalchemy.Text, nullable=False) - status = sqlalchemy.Column(sqlalchemy.Text, nullable=False) - version = sqlalchemy.Column(sqlalchemy.Text, nullable=False) - severity = sqlalchemy.Column(sqlalchemy.Text, nullable=False) - rights = sqlalchemy.Column(sqlalchemy.Text, nullable=False) + last_release_log: Mapped[Optional[str]] = mapped_column( + sqlalchemy.Text, nullable=True + ) + summary: Mapped[Optional[str]] = mapped_column( + sqlalchemy.Text, nullable=True + ) + solution: Mapped[Optional[str]] = mapped_column( + sqlalchemy.Text, nullable=True + ) + + freezed: Mapped[Optional[bool]] = mapped_column( + sqlalchemy.Boolean, nullable=True + ) + + issued_date: Mapped[datetime.datetime] = mapped_column( + sqlalchemy.DateTime, nullable=False + ) + updated_date: Mapped[datetime.datetime] = mapped_column( + sqlalchemy.DateTime, nullable=False + ) + description: Mapped[Optional[str]] = mapped_column( + sqlalchemy.Text, nullable=True + ) + original_description: Mapped[str] = mapped_column( + sqlalchemy.Text, nullable=False + ) + title: Mapped[Optional[str]] = mapped_column( + sqlalchemy.Text, nullable=True + ) + oval_title: Mapped[Optional[str]] = mapped_column( + sqlalchemy.Text, nullable=True + ) + original_title: Mapped[str] = mapped_column( + sqlalchemy.Text, nullable=False + ) + contact_mail: Mapped[str] = mapped_column(sqlalchemy.Text, nullable=False) + status: Mapped[str] = mapped_column(sqlalchemy.Text, nullable=False) + version: Mapped[str] = mapped_column(sqlalchemy.Text, nullable=False) + severity: Mapped[str] = mapped_column(sqlalchemy.Text, nullable=False) + rights: Mapped[str] = mapped_column(sqlalchemy.Text, nullable=False) # OVAL-only fields - definition_id = sqlalchemy.Column(sqlalchemy.Text, nullable=False) - definition_version = sqlalchemy.Column(sqlalchemy.Text, nullable=False) - definition_class = sqlalchemy.Column(sqlalchemy.Text, nullable=False) - affected_cpe = sqlalchemy.Column(JSONB, nullable=False, default=[]) - criteria = sqlalchemy.Column(JSONB, nullable=True) - original_criteria = sqlalchemy.Column(JSONB, nullable=True) - tests = sqlalchemy.Column(JSONB, nullable=True) - original_tests = sqlalchemy.Column(JSONB, nullable=True) - objects = sqlalchemy.Column(JSONB, nullable=True) - original_objects = sqlalchemy.Column(JSONB, nullable=True) - states = sqlalchemy.Column(JSONB, nullable=True) - original_states = sqlalchemy.Column(JSONB, nullable=True) - variables = sqlalchemy.Column(JSONB, nullable=True) - original_variables = sqlalchemy.Column(JSONB, nullable=True) - - references = relationship( + definition_id: Mapped[str] = mapped_column(sqlalchemy.Text, nullable=False) + definition_version: Mapped[str] = mapped_column( + sqlalchemy.Text, nullable=False + ) + definition_class: Mapped[str] = mapped_column( + sqlalchemy.Text, nullable=False + ) + affected_cpe: Mapped[Dict[str, Any]] = mapped_column( + JSONB, nullable=False, default=[] + ) + criteria: Mapped[Optional[Dict[str, Any]]] = mapped_column( + JSONB, nullable=True + ) + original_criteria: Mapped[Optional[Dict[str, Any]]] = mapped_column( + JSONB, nullable=True + ) + tests: Mapped[Optional[Dict[str, Any]]] = mapped_column( + JSONB, nullable=True + ) + original_tests: Mapped[Optional[Dict[str, Any]]] = mapped_column( + JSONB, nullable=True + ) + objects: Mapped[Optional[Dict[str, Any]]] = mapped_column( + JSONB, nullable=True + ) + original_objects: Mapped[Optional[Dict[str, Any]]] = mapped_column( + JSONB, nullable=True + ) + states: Mapped[Optional[Dict[str, Any]]] = mapped_column( + JSONB, nullable=True + ) + original_states: Mapped[Optional[Dict[str, Any]]] = mapped_column( + JSONB, nullable=True + ) + variables: Mapped[Optional[Dict[str, Any]]] = mapped_column( + JSONB, nullable=True + ) + original_variables: Mapped[Optional[Dict[str, Any]]] = mapped_column( + JSONB, nullable=True + ) + + references: Mapped[List["NewErrataReference"]] = relationship( "NewErrataReference", back_populates="platform_specific_errata_record", ) - packages = relationship( + packages: Mapped[List["NewErrataPackage"]] = relationship( "NewErrataPackage", back_populates="platform_specific_errata_record", ) - cves = association_proxy("references", "cve_id") + cves: Mapped[AssociationProxy[Any]] = association_proxy( + "references", "cve_id" + ) def get_description(self): if self.description: @@ -1314,10 +1525,10 @@ class NewErrataPackage(Base): ), ) - id = sqlalchemy.Column(sqlalchemy.Integer, primary_key=True) - errata_record_id = sqlalchemy.Column(sqlalchemy.Text) - platform_id = sqlalchemy.Column(sqlalchemy.Integer) - platform_specific_errata_record = relationship( + id: Mapped[int] = mapped_column(sqlalchemy.Integer, primary_key=True) + errata_record_id: Mapped[str] = mapped_column(sqlalchemy.Text) + platform_id: Mapped[int] = mapped_column(sqlalchemy.Integer) + platform_specific_errata_record: Mapped["NewErrataRecord"] = relationship( "NewErrataRecord", foreign_keys=[errata_record_id, platform_id], cascade="all, delete", @@ -1325,14 +1536,18 @@ class NewErrataPackage(Base): "NewErrataPackage.platform_id == NewErrataRecord.platform_id)", back_populates="packages", ) - name = sqlalchemy.Column(sqlalchemy.Text, nullable=False) - version = sqlalchemy.Column(sqlalchemy.Text, nullable=False) - release = sqlalchemy.Column(sqlalchemy.Text, nullable=False) - epoch = sqlalchemy.Column(sqlalchemy.Integer, nullable=False) - arch = sqlalchemy.Column(sqlalchemy.Text, nullable=False) - source_srpm = sqlalchemy.Column(sqlalchemy.Text, nullable=True) - reboot_suggested = sqlalchemy.Column(sqlalchemy.Boolean, nullable=False) - albs_packages = relationship( + name: Mapped[str] = mapped_column(sqlalchemy.Text, nullable=False) + version: Mapped[str] = mapped_column(sqlalchemy.Text, nullable=False) + release: Mapped[str] = mapped_column(sqlalchemy.Text, nullable=False) + epoch: Mapped[int] = mapped_column(sqlalchemy.Integer, nullable=False) + arch: Mapped[str] = mapped_column(sqlalchemy.Text, nullable=False) + source_srpm: Mapped[Optional[str]] = mapped_column( + sqlalchemy.Text, nullable=True + ) + reboot_suggested: Mapped[bool] = mapped_column( + sqlalchemy.Boolean, nullable=False + ) + albs_packages: Mapped[List["NewErrataToALBSPackage"]] = relationship( "NewErrataToALBSPackage", back_populates="errata_package", cascade="all, delete", @@ -1349,16 +1564,21 @@ class NewErrataReference(Base): ), ) - id = sqlalchemy.Column(sqlalchemy.Integer, primary_key=True) - href = sqlalchemy.Column(sqlalchemy.Text, nullable=False) - ref_id = sqlalchemy.Column(sqlalchemy.Text, nullable=False) - title = sqlalchemy.Column(sqlalchemy.Text, nullable=False) - ref_type = sqlalchemy.Column( - sqlalchemy.Enum(ErrataReferenceType), nullable=False - ) - errata_record_id = sqlalchemy.Column(sqlalchemy.Text) - platform_id = sqlalchemy.Column(sqlalchemy.Integer) - platform_specific_errata_record = relationship( + id: Mapped[int] = mapped_column(sqlalchemy.Integer, primary_key=True) + href: Mapped[str] = mapped_column(sqlalchemy.Text, nullable=False) + ref_id: Mapped[str] = mapped_column(sqlalchemy.Text, nullable=False) + title: Mapped[str] = mapped_column(sqlalchemy.Text, nullable=False) + ref_type: Mapped[ + Literal[ + ErrataReferenceType.cve, + ErrataReferenceType.rhsa, + ErrataReferenceType.self_ref, + ErrataReferenceType.bugzilla, + ] + ] = mapped_column(sqlalchemy.Enum(ErrataReferenceType), nullable=False) + errata_record_id: Mapped[str] = mapped_column(sqlalchemy.Text) + platform_id: Mapped[int] = mapped_column(sqlalchemy.Integer) + platform_specific_errata_record: Mapped["NewErrataRecord"] = relationship( "NewErrataRecord", foreign_keys=[errata_record_id, platform_id], cascade="all, delete", @@ -1366,8 +1586,8 @@ class NewErrataReference(Base): "NewErrataReference.platform_id == NewErrataRecord.platform_id)", back_populates="references", ) - cve = relationship("ErrataCVE", cascade="all, delete") - cve_id = sqlalchemy.Column( + cve: Mapped["ErrataCVE"] = relationship("ErrataCVE", cascade="all, delete") + cve_id: Mapped[Optional[str]] = mapped_column( sqlalchemy.Text, sqlalchemy.ForeignKey( "errata_cves.id", @@ -1387,8 +1607,8 @@ class NewErrataToALBSPackage(Base): ), ) - id = sqlalchemy.Column(sqlalchemy.Integer, primary_key=True) - errata_package_id = sqlalchemy.Column( + id: Mapped[int] = mapped_column(sqlalchemy.Integer, primary_key=True) + errata_package_id: Mapped[int] = mapped_column( sqlalchemy.Integer, sqlalchemy.ForeignKey( "new_errata_packages.id", @@ -1398,11 +1618,11 @@ class NewErrataToALBSPackage(Base): nullable=False, index=True, ) - errata_package = relationship( + errata_package: Mapped[Optional[int]] = relationship( "NewErrataPackage", back_populates="albs_packages", ) - albs_artifact_id = sqlalchemy.Column( + albs_artifact_id = mapped_column( sqlalchemy.Integer, sqlalchemy.ForeignKey( "build_artifacts.id", @@ -1411,18 +1631,29 @@ class NewErrataToALBSPackage(Base): ), nullable=True, ) - build_artifact: BuildTaskArtifact = relationship("BuildTaskArtifact") - pulp_href = sqlalchemy.Column(sqlalchemy.Text, nullable=True) - status = sqlalchemy.Column( + build_artifact: Mapped["BuildTaskArtifact"] = relationship( + "BuildTaskArtifact" + ) + pulp_href: Mapped[Optional[str]] = mapped_column( + sqlalchemy.Text, nullable=True + ) + status: Mapped[ + Literal[ + ErrataPackageStatus.proposal, + ErrataPackageStatus.skipped, + ErrataPackageStatus.released, + ErrataPackageStatus.approved, + ] + ] = mapped_column( sqlalchemy.Enum(ErrataPackageStatus), nullable=False, ) - name = sqlalchemy.Column(sqlalchemy.Text, nullable=False) - arch = sqlalchemy.Column(sqlalchemy.Text, nullable=False) - version = sqlalchemy.Column(sqlalchemy.Text, nullable=False) - release = sqlalchemy.Column(sqlalchemy.Text, nullable=False) - epoch = sqlalchemy.Column(sqlalchemy.Integer, nullable=False) + name: Mapped[str] = mapped_column(sqlalchemy.Text, nullable=False) + arch: Mapped[str] = mapped_column(sqlalchemy.Text, nullable=False) + version: Mapped[str] = mapped_column(sqlalchemy.Text, nullable=False) + release: Mapped[str] = mapped_column(sqlalchemy.Text, nullable=False) + epoch: Mapped[int] = mapped_column(sqlalchemy.Integer, nullable=False) @property def build_id(self): @@ -1444,56 +1675,121 @@ def get_pulp_href(self): class ErrataRecord(Base): __tablename__ = "errata_records" - id = sqlalchemy.Column(sqlalchemy.Text, primary_key=True) - platform_id = sqlalchemy.Column( + id: Mapped[str] = mapped_column(sqlalchemy.Text, primary_key=True) + platform_id: Mapped[int] = mapped_column( sqlalchemy.Integer, sqlalchemy.ForeignKey("platforms.id"), nullable=False, ) - platform = relationship("Platform") - module = sqlalchemy.Column(sqlalchemy.Text, nullable=True) - release_status = sqlalchemy.Column( + platform: Mapped["Platform"] = relationship("Platform") + module: Mapped[Optional[str]] = mapped_column( + sqlalchemy.Text, nullable=True + ) + release_status: Mapped[ + Optional[ + Literal[ + ErrataReleaseStatus.NOT_RELEASED, + ErrataReleaseStatus.IN_PROGRESS, + ErrataReleaseStatus.RELEASED, + ErrataReleaseStatus.FAILED, + ] + ] + ] = mapped_column( sqlalchemy.Enum(ErrataReleaseStatus), nullable=True, ) - last_release_log = sqlalchemy.Column(sqlalchemy.Text, nullable=True) - summary = sqlalchemy.Column(sqlalchemy.Text, nullable=True) - solution = sqlalchemy.Column(sqlalchemy.Text, nullable=True) - - freezed = sqlalchemy.Column(sqlalchemy.Boolean, nullable=True) - - issued_date = sqlalchemy.Column(sqlalchemy.DateTime, nullable=False) - updated_date = sqlalchemy.Column(sqlalchemy.DateTime, nullable=False) - description = sqlalchemy.Column(sqlalchemy.Text, nullable=True) - original_description = sqlalchemy.Column(sqlalchemy.Text, nullable=False) - title = sqlalchemy.Column(sqlalchemy.Text, nullable=True) - oval_title = sqlalchemy.Column(sqlalchemy.Text, nullable=True) - original_title = sqlalchemy.Column(sqlalchemy.Text, nullable=False) - contact_mail = sqlalchemy.Column(sqlalchemy.Text, nullable=False) - status = sqlalchemy.Column(sqlalchemy.Text, nullable=False) - version = sqlalchemy.Column(sqlalchemy.Text, nullable=False) - severity = sqlalchemy.Column(sqlalchemy.Text, nullable=False) - rights = sqlalchemy.Column(sqlalchemy.Text, nullable=False) + last_release_log: Mapped[Optional[str]] = mapped_column( + sqlalchemy.Text, nullable=True + ) + summary: Mapped[Optional[str]] = mapped_column( + sqlalchemy.Text, nullable=True + ) + solution: Mapped[Optional[str]] = mapped_column( + sqlalchemy.Text, nullable=True + ) + + freezed: Mapped[Optional[bool]] = mapped_column( + sqlalchemy.Boolean, nullable=True + ) + + issued_date: Mapped[datetime.datetime] = mapped_column( + sqlalchemy.DateTime, nullable=False + ) + updated_date: Mapped[datetime.datetime] = mapped_column( + sqlalchemy.DateTime, nullable=False + ) + description: Mapped[Optional[str]] = mapped_column( + sqlalchemy.Text, nullable=True + ) + original_description: Mapped[str] = mapped_column( + sqlalchemy.Text, nullable=False + ) + title: Mapped[Optional[str]] = mapped_column( + sqlalchemy.Text, nullable=True + ) + oval_title: Mapped[Optional[str]] = mapped_column( + sqlalchemy.Text, nullable=True + ) + original_title: Mapped[str] = mapped_column( + sqlalchemy.Text, nullable=False + ) + contact_mail: Mapped[str] = mapped_column(sqlalchemy.Text, nullable=False) + status: Mapped[str] = mapped_column(sqlalchemy.Text, nullable=False) + version: Mapped[str] = mapped_column(sqlalchemy.Text, nullable=False) + severity: Mapped[str] = mapped_column(sqlalchemy.Text, nullable=False) + rights: Mapped[str] = mapped_column(sqlalchemy.Text, nullable=False) # OVAL-only fields - definition_id = sqlalchemy.Column(sqlalchemy.Text, nullable=False) - definition_version = sqlalchemy.Column(sqlalchemy.Text, nullable=False) - definition_class = sqlalchemy.Column(sqlalchemy.Text, nullable=False) - affected_cpe = sqlalchemy.Column(JSONB, nullable=False, default=[]) - criteria = sqlalchemy.Column(JSONB, nullable=True) - original_criteria = sqlalchemy.Column(JSONB, nullable=True) - tests = sqlalchemy.Column(JSONB, nullable=True) - original_tests = sqlalchemy.Column(JSONB, nullable=True) - objects = sqlalchemy.Column(JSONB, nullable=True) - original_objects = sqlalchemy.Column(JSONB, nullable=True) - states = sqlalchemy.Column(JSONB, nullable=True) - original_states = sqlalchemy.Column(JSONB, nullable=True) - variables = sqlalchemy.Column(JSONB, nullable=True) - original_variables = sqlalchemy.Column(JSONB, nullable=True) - - references = relationship("ErrataReference", cascade="all, delete") - packages = relationship("ErrataPackage", cascade="all, delete") - - cves = association_proxy("references", "cve_id") + definition_id: Mapped[str] = mapped_column(sqlalchemy.Text, nullable=False) + definition_version: Mapped[str] = mapped_column( + sqlalchemy.Text, nullable=False + ) + definition_class: Mapped[str] = mapped_column( + sqlalchemy.Text, nullable=False + ) + affected_cpe: Mapped[Dict[str, Any]] = mapped_column( + JSONB, nullable=False, default=[] + ) + criteria: Mapped[Optional[Dict[str, Any]]] = mapped_column( + JSONB, nullable=True + ) + original_criteria: Mapped[Optional[Dict[str, Any]]] = mapped_column( + JSONB, nullable=True + ) + tests: Mapped[Optional[Dict[str, Any]]] = mapped_column( + JSONB, nullable=True + ) + original_tests: Mapped[Optional[Dict[str, Any]]] = mapped_column( + JSONB, nullable=True + ) + objects: Mapped[Optional[Dict[str, Any]]] = mapped_column( + JSONB, nullable=True + ) + original_objects: Mapped[Optional[Dict[str, Any]]] = mapped_column( + JSONB, nullable=True + ) + states: Mapped[Optional[Dict[str, Any]]] = mapped_column( + JSONB, nullable=True + ) + original_states: Mapped[Optional[Dict[str, Any]]] = mapped_column( + JSONB, nullable=True + ) + variables: Mapped[Optional[Dict[str, Any]]] = mapped_column( + JSONB, nullable=True + ) + original_variables: Mapped[Optional[Dict[str, Any]]] = mapped_column( + JSONB, nullable=True + ) + + references: Mapped[List["ErrataReference"]] = relationship( + "ErrataReference", cascade="all, delete" + ) + packages: Mapped[List["ErrataPackage"]] = relationship( + "ErrataPackage", cascade="all, delete" + ) + + cves: Mapped[AssociationProxy[Any]] = association_proxy( + "references", "cve_id" + ) def get_description(self): if self.description: @@ -1520,14 +1816,19 @@ def get_type(self): class ErrataReference(Base): __tablename__ = "errata_references" - id = sqlalchemy.Column(sqlalchemy.Integer, primary_key=True) - href = sqlalchemy.Column(sqlalchemy.Text, nullable=False) - ref_id = sqlalchemy.Column(sqlalchemy.Text, nullable=False) - title = sqlalchemy.Column(sqlalchemy.Text, nullable=False) - ref_type = sqlalchemy.Column( - sqlalchemy.Enum(ErrataReferenceType), nullable=False - ) - errata_record_id = sqlalchemy.Column( + id: Mapped[int] = mapped_column(sqlalchemy.Integer, primary_key=True) + href: Mapped[str] = mapped_column(sqlalchemy.Text, nullable=False) + ref_id: Mapped[str] = mapped_column(sqlalchemy.Text, nullable=False) + title: Mapped[str] = mapped_column(sqlalchemy.Text, nullable=False) + ref_type: Mapped[ + Literal[ + ErrataReferenceType.cve, + ErrataReferenceType.rhsa, + ErrataReferenceType.self_ref, + ErrataReferenceType.bugzilla, + ] + ] = mapped_column(sqlalchemy.Enum(ErrataReferenceType), nullable=False) + errata_record_id: Mapped[str] = mapped_column( sqlalchemy.Text, sqlalchemy.ForeignKey( "errata_records.id", @@ -1537,8 +1838,8 @@ class ErrataReference(Base): nullable=False, index=True, ) - cve = relationship("ErrataCVE", cascade="all, delete") - cve_id = sqlalchemy.Column( + cve: Mapped["ErrataCVE"] = relationship("ErrataCVE", cascade="all, delete") + cve_id: Mapped[Optional[str]] = mapped_column( sqlalchemy.Text, sqlalchemy.ForeignKey( "errata_cves.id", @@ -1552,18 +1853,20 @@ class ErrataReference(Base): class ErrataCVE(Base): __tablename__ = "errata_cves" - id = sqlalchemy.Column(sqlalchemy.Text, primary_key=True) - cvss3 = sqlalchemy.Column(sqlalchemy.Text, nullable=False) - cwe = sqlalchemy.Column(sqlalchemy.Text, nullable=True) - impact = sqlalchemy.Column(sqlalchemy.Text, nullable=True) - public = sqlalchemy.Column(sqlalchemy.Text, nullable=False) + id: Mapped[str] = mapped_column(sqlalchemy.Text, primary_key=True) + cvss3: Mapped[str] = mapped_column(sqlalchemy.Text, nullable=False) + cwe: Mapped[Optional[str]] = mapped_column(sqlalchemy.Text, nullable=True) + impact: Mapped[Optional[str]] = mapped_column( + sqlalchemy.Text, nullable=True + ) + public: Mapped[str] = mapped_column(sqlalchemy.Text, nullable=False) class ErrataPackage(Base): __tablename__ = "errata_packages" - id = sqlalchemy.Column(sqlalchemy.Integer, primary_key=True) - errata_record_id = sqlalchemy.Column( + id: Mapped[int] = mapped_column(sqlalchemy.Integer, primary_key=True) + errata_record_id: Mapped[str] = mapped_column( sqlalchemy.Text, sqlalchemy.ForeignKey( "errata_records.id", @@ -1573,14 +1876,18 @@ class ErrataPackage(Base): nullable=False, index=True, ) - name = sqlalchemy.Column(sqlalchemy.Text, nullable=False) - version = sqlalchemy.Column(sqlalchemy.Text, nullable=False) - release = sqlalchemy.Column(sqlalchemy.Text, nullable=False) - epoch = sqlalchemy.Column(sqlalchemy.Integer, nullable=False) - arch = sqlalchemy.Column(sqlalchemy.Text, nullable=False) - source_srpm = sqlalchemy.Column(sqlalchemy.Text, nullable=True) - reboot_suggested = sqlalchemy.Column(sqlalchemy.Boolean, nullable=False) - albs_packages = relationship( + name: Mapped[str] = mapped_column(sqlalchemy.Text, nullable=False) + version: Mapped[str] = mapped_column(sqlalchemy.Text, nullable=False) + release: Mapped[str] = mapped_column(sqlalchemy.Text, nullable=False) + epoch: Mapped[int] = mapped_column(sqlalchemy.Integer, nullable=False) + arch: Mapped[str] = mapped_column(sqlalchemy.Text, nullable=False) + source_srpm: Mapped[Optional[str]] = mapped_column( + sqlalchemy.Text, nullable=True + ) + reboot_suggested: Mapped[bool] = mapped_column( + sqlalchemy.Boolean, nullable=False + ) + albs_packages: Mapped[List["ErrataToALBSPackage"]] = relationship( "ErrataToALBSPackage", back_populates="errata_package", cascade="all, delete", @@ -1596,8 +1903,8 @@ class ErrataToALBSPackage(Base): ), ) - id = sqlalchemy.Column(sqlalchemy.Integer, primary_key=True) - errata_package_id = sqlalchemy.Column( + id: Mapped[int] = mapped_column(sqlalchemy.Integer, primary_key=True) + errata_package_id: Mapped[int] = mapped_column( sqlalchemy.Integer, sqlalchemy.ForeignKey( "errata_packages.id", @@ -1607,11 +1914,11 @@ class ErrataToALBSPackage(Base): nullable=False, index=True, ) - errata_package = relationship( + errata_package: Mapped["ErrataPackage"] = relationship( "ErrataPackage", back_populates="albs_packages", ) - albs_artifact_id = sqlalchemy.Column( + albs_artifact_id: Mapped[Optional[int]] = mapped_column( sqlalchemy.Integer, sqlalchemy.ForeignKey( "build_artifacts.id", @@ -1620,18 +1927,29 @@ class ErrataToALBSPackage(Base): ), nullable=True, ) - build_artifact: BuildTaskArtifact = relationship("BuildTaskArtifact") - pulp_href = sqlalchemy.Column(sqlalchemy.Text, nullable=True) - status = sqlalchemy.Column( + build_artifact: Mapped["BuildTaskArtifact"] = relationship( + "BuildTaskArtifact" + ) + pulp_href: Mapped[Optional[str]] = mapped_column( + sqlalchemy.Text, nullable=True + ) + status: Mapped[ + Literal[ + ErrataPackageStatus.proposal, + ErrataPackageStatus.skipped, + ErrataPackageStatus.released, + ErrataPackageStatus.approved, + ] + ] = mapped_column( sqlalchemy.Enum(ErrataPackageStatus), nullable=False, ) - name = sqlalchemy.Column(sqlalchemy.Text, nullable=False) - arch = sqlalchemy.Column(sqlalchemy.Text, nullable=False) - version = sqlalchemy.Column(sqlalchemy.Text, nullable=False) - release = sqlalchemy.Column(sqlalchemy.Text, nullable=False) - epoch = sqlalchemy.Column(sqlalchemy.Integer, nullable=False) + name: Mapped[str] = mapped_column(sqlalchemy.Text, nullable=False) + arch: Mapped[str] = mapped_column(sqlalchemy.Text, nullable=False) + version: Mapped[str] = mapped_column(sqlalchemy.Text, nullable=False) + release: Mapped[str] = mapped_column(sqlalchemy.Text, nullable=False) + epoch: Mapped[int] = mapped_column(sqlalchemy.Integer, nullable=False) @property def build_id(self): @@ -1652,12 +1970,12 @@ def get_pulp_href(self): class PerformanceStats(Base): __tablename__ = "performance_stats" - id: int = sqlalchemy.Column(sqlalchemy.Integer, primary_key=True) - statistics: Dict[str, Dict[str, Dict[str, str]]] = sqlalchemy.Column( + id: Mapped[int] = mapped_column(sqlalchemy.Integer, primary_key=True) + statistics: Mapped[Optional[Dict[str, Any]]] = mapped_column( JSONB, nullable=True, ) - build_task_id: int = sqlalchemy.Column( + build_task_id: Mapped[Optional[int]] = mapped_column( sqlalchemy.Integer, sqlalchemy.ForeignKey( "build_tasks.id", @@ -1666,21 +1984,21 @@ class PerformanceStats(Base): nullable=True, index=True, ) - build_task: BuildTask = relationship( + build_task: Mapped["BuildTask"] = relationship( "BuildTask", back_populates="performance_stats", ) - test_task_id: int = sqlalchemy.Column( + test_task_id: Mapped[Optional[int]] = mapped_column( sqlalchemy.Integer, sqlalchemy.ForeignKey("test_tasks.id", name="perf_stats_test_task_id"), nullable=True, index=True, ) - test_task: TestTask = relationship( + test_task: Mapped["TestTask"] = relationship( "TestTask", back_populates="performance_stats", ) - release_id: int = sqlalchemy.Column( + release_id: Mapped[Optional[int]] = mapped_column( sqlalchemy.Integer, sqlalchemy.ForeignKey( "build_releases.id", @@ -1689,7 +2007,7 @@ class PerformanceStats(Base): nullable=True, index=True, ) - release: Release = relationship( + release: Mapped["Release"] = relationship( "Release", back_populates="performance_stats", ) diff --git a/alws/pulp_models.py b/alws/pulp_models.py index 207907adb..e3c9e5c2b 100644 --- a/alws/pulp_models.py +++ b/alws/pulp_models.py @@ -1,10 +1,10 @@ +import datetime import uuid -from datetime import datetime -from typing import Any, Dict, List +from typing import Any, Dict, List, Optional import sqlalchemy from sqlalchemy.dialects.postgresql import JSONB, UUID -from sqlalchemy.orm import relationship +from sqlalchemy.orm import Mapped, mapped_column, relationship from alws.database import PulpBase @@ -12,96 +12,124 @@ class UpdateRecord(PulpBase): __tablename__ = "rpm_updaterecord" - content_ptr_id = sqlalchemy.Column(UUID(as_uuid=True), primary_key=True) - id = sqlalchemy.Column(sqlalchemy.Text) - issued_date = sqlalchemy.Column(sqlalchemy.Text) - updated_date = sqlalchemy.Column(sqlalchemy.Text) - description = sqlalchemy.Column(sqlalchemy.Text) - fromstr = sqlalchemy.Column(sqlalchemy.Text) - status = sqlalchemy.Column(sqlalchemy.Text) - title = sqlalchemy.Column(sqlalchemy.Text) - summary = sqlalchemy.Column(sqlalchemy.Text) - version = sqlalchemy.Column(sqlalchemy.Text) - type = sqlalchemy.Column(sqlalchemy.Text) - severity = sqlalchemy.Column(sqlalchemy.Text) - solution = sqlalchemy.Column(sqlalchemy.Text) - release = sqlalchemy.Column(sqlalchemy.Text) - rights = sqlalchemy.Column(sqlalchemy.Text) - pushcount = sqlalchemy.Column(sqlalchemy.Text) - digest = sqlalchemy.Column(sqlalchemy.Text) - reboot_suggested = sqlalchemy.Column(sqlalchemy.Boolean) - - collections: List["UpdateCollection"] = relationship("UpdateCollection") - references: List["UpdateReference"] = relationship("UpdateReference") + content_ptr_id: Mapped[uuid.UUID] = mapped_column( + UUID(as_uuid=True), primary_key=True + ) + id: Mapped[str] = mapped_column(sqlalchemy.Text) + issued_date: Mapped[str] = mapped_column(sqlalchemy.Text) + updated_date: Mapped[str] = mapped_column(sqlalchemy.Text) + description: Mapped[str] = mapped_column(sqlalchemy.Text) + fromstr: Mapped[str] = mapped_column(sqlalchemy.Text) + status: Mapped[str] = mapped_column(sqlalchemy.Text) + title: Mapped[str] = mapped_column(sqlalchemy.Text) + summary: Mapped[str] = mapped_column(sqlalchemy.Text) + version: Mapped[str] = mapped_column(sqlalchemy.Text) + type: Mapped[str] = mapped_column(sqlalchemy.Text) + severity: Mapped[str] = mapped_column(sqlalchemy.Text) + solution: Mapped[str] = mapped_column(sqlalchemy.Text) + release: Mapped[str] = mapped_column(sqlalchemy.Text) + rights: Mapped[str] = mapped_column(sqlalchemy.Text) + pushcount: Mapped[str] = mapped_column(sqlalchemy.Text) + digest: Mapped[str] = mapped_column(sqlalchemy.Text) + reboot_suggested: Mapped[bool] = mapped_column(sqlalchemy.Boolean) + + collections: Mapped[List["UpdateCollection"]] = relationship( + "UpdateCollection" + ) + references: Mapped[List["UpdateReference"]] = relationship( + "UpdateReference" + ) class UpdateCollection(PulpBase): __tablename__ = "rpm_updatecollection" - pulp_id = sqlalchemy.Column(UUID(as_uuid=True), primary_key=True) - pulp_created = sqlalchemy.Column(sqlalchemy.DATETIME) - pulp_last_updated = sqlalchemy.Column(sqlalchemy.DATETIME) + pulp_id: Mapped[uuid.UUID] = mapped_column( + UUID(as_uuid=True), primary_key=True + ) + pulp_created: Mapped[datetime.datetime] = mapped_column( + sqlalchemy.DATETIME + ) + pulp_last_updated: Mapped[datetime.datetime] = mapped_column( + sqlalchemy.DATETIME + ) - name = sqlalchemy.Column(sqlalchemy.Text) - shortname = sqlalchemy.Column(sqlalchemy.Text) - module = sqlalchemy.Column(sqlalchemy.JSON, nullable=True) + name: Mapped[str] = mapped_column(sqlalchemy.Text) + shortname: Mapped[str] = mapped_column(sqlalchemy.Text) + module: Mapped[Optional[Dict[str, Any]]] = mapped_column( + sqlalchemy.JSON, nullable=True + ) - update_record_id = sqlalchemy.Column( + update_record_id: Mapped[uuid.UUID] = mapped_column( UUID(as_uuid=True), sqlalchemy.ForeignKey(UpdateRecord.content_ptr_id), nullable=False, ) - packages: List["UpdatePackage"] = relationship("UpdatePackage") + packages: Mapped[List["UpdatePackage"]] = relationship("UpdatePackage") class UpdatePackage(PulpBase): __tablename__ = "rpm_updatecollectionpackage" - pulp_id = sqlalchemy.Column( + pulp_id: Mapped[uuid.UUID] = mapped_column( UUID(as_uuid=True), primary_key=True, default=uuid.uuid4 ) - pulp_created = sqlalchemy.Column(sqlalchemy.DATETIME, default=datetime.now) - pulp_last_updated = sqlalchemy.Column( + pulp_created: Mapped[datetime.datetime] = mapped_column( + sqlalchemy.DATETIME, default=datetime.datetime.now + ) + pulp_last_updated: Mapped[datetime.datetime] = mapped_column( sqlalchemy.DATETIME, - default=datetime.now, - ) - arch = sqlalchemy.Column(sqlalchemy.Text) - filename = sqlalchemy.Column(sqlalchemy.Text) - name = sqlalchemy.Column(sqlalchemy.Text) - version = sqlalchemy.Column(sqlalchemy.Text) - release = sqlalchemy.Column(sqlalchemy.Text) - epoch = sqlalchemy.Column(sqlalchemy.Text) - reboot_suggested = sqlalchemy.Column(sqlalchemy.Boolean, default=False) - relogin_suggested = sqlalchemy.Column(sqlalchemy.Boolean, default=False) - restart_suggested = sqlalchemy.Column(sqlalchemy.Boolean, default=False) - src = sqlalchemy.Column(sqlalchemy.Text) - sum = sqlalchemy.Column(sqlalchemy.Text) - update_collection_id = sqlalchemy.Column( + default=datetime.datetime.now, + ) + arch: Mapped[str] = mapped_column(sqlalchemy.Text) + filename: Mapped[str] = mapped_column(sqlalchemy.Text) + name: Mapped[str] = mapped_column(sqlalchemy.Text) + version: Mapped[str] = mapped_column(sqlalchemy.Text) + release: Mapped[str] = mapped_column(sqlalchemy.Text) + epoch: Mapped[str] = mapped_column(sqlalchemy.Text) + reboot_suggested: Mapped[bool] = mapped_column( + sqlalchemy.Boolean, default=False + ) + relogin_suggested: Mapped[bool] = mapped_column( + sqlalchemy.Boolean, default=False + ) + restart_suggested: Mapped[bool] = mapped_column( + sqlalchemy.Boolean, default=False + ) + src: Mapped[str] = mapped_column(sqlalchemy.Text) + sum: Mapped[str] = mapped_column(sqlalchemy.Text) + update_collection_id: Mapped[uuid.UUID] = mapped_column( UUID(as_uuid=True), sqlalchemy.ForeignKey(UpdateCollection.pulp_id), nullable=False, ) - sum_type = sqlalchemy.Column(sqlalchemy.Integer) + sum_type: Mapped[int] = mapped_column(sqlalchemy.Integer) class UpdateReference(PulpBase): __tablename__ = "rpm_updatereference" - pulp_id = sqlalchemy.Column( + pulp_id: Mapped[uuid.UUID] = mapped_column( UUID(as_uuid=True), primary_key=True, default=uuid.uuid4, ) - pulp_created = sqlalchemy.Column(sqlalchemy.DATETIME, default=datetime.now) - pulp_last_updated = sqlalchemy.Column( + pulp_created: Mapped[datetime.datetime] = mapped_column( + sqlalchemy.DATETIME, default=datetime.datetime.now + ) + pulp_last_updated: Mapped[datetime.datetime] = mapped_column( sqlalchemy.DATETIME, - default=datetime.now, + default=datetime.datetime.now, + ) + href: Mapped[str] = mapped_column(sqlalchemy.Text) + ref_id: Mapped[Optional[str]] = mapped_column( + sqlalchemy.Text, nullable=True ) - href = sqlalchemy.Column(sqlalchemy.Text) - ref_id = sqlalchemy.Column(sqlalchemy.Text, nullable=True) - title = sqlalchemy.Column(sqlalchemy.Text, nullable=True) - ref_type = sqlalchemy.Column(sqlalchemy.Text) - update_record_id = sqlalchemy.Column( + title: Mapped[Optional[str]] = mapped_column( + sqlalchemy.Text, nullable=True + ) + ref_type: Mapped[str] = mapped_column(sqlalchemy.Text) + update_record_id: Mapped[uuid.UUID] = mapped_column( UUID(as_uuid=True), sqlalchemy.ForeignKey(UpdateRecord.content_ptr_id), nullable=False, @@ -111,29 +139,37 @@ class UpdateReference(PulpBase): class CoreRepository(PulpBase): __tablename__ = "core_repository" - pulp_id = sqlalchemy.Column( + pulp_id: Mapped[uuid.UUID] = mapped_column( UUID(as_uuid=True), primary_key=True, default=uuid.uuid4, ) - pulp_created = sqlalchemy.Column(sqlalchemy.DATETIME, default=datetime.now) - pulp_last_updated = sqlalchemy.Column( + pulp_created: Mapped[datetime.datetime] = mapped_column( + sqlalchemy.DATETIME, default=datetime.datetime.now + ) + pulp_last_updated: Mapped[datetime.datetime] = mapped_column( sqlalchemy.DATETIME, - default=datetime.now, + default=datetime.datetime.now, + ) + name: Mapped[str] = mapped_column(sqlalchemy.Text) + description: Mapped[Optional[str]] = mapped_column( + sqlalchemy.Text, nullable=True ) - name = sqlalchemy.Column(sqlalchemy.Text) - description = sqlalchemy.Column(sqlalchemy.Text, nullable=True) - next_version = sqlalchemy.Column(sqlalchemy.Integer) - pulp_type = sqlalchemy.Column(sqlalchemy.Text) - remote_id = sqlalchemy.Column(UUID(as_uuid=True), nullable=True) - retain_repo_versions = sqlalchemy.Column(sqlalchemy.Integer, nullable=True) - user_hidden = sqlalchemy.Column(sqlalchemy.Boolean) + next_version: Mapped[int] = mapped_column(sqlalchemy.Integer) + pulp_type: Mapped[str] = mapped_column(sqlalchemy.Text) + remote_id: Mapped[Optional[uuid.UUID]] = mapped_column( + UUID(as_uuid=True), nullable=True + ) + retain_repo_versions: Mapped[Optional[int]] = mapped_column( + sqlalchemy.Integer, nullable=True + ) + user_hidden: Mapped[bool] = mapped_column(sqlalchemy.Boolean) - repository_content = relationship( + repository_content: Mapped["CoreRepositoryContent"] = relationship( "CoreRepositoryContent", back_populates="repository", ) - versions = relationship( + versions: Mapped[List["CoreRepositoryVersion"]] = relationship( "CoreRepositoryVersion", back_populates="repository", ) @@ -142,26 +178,30 @@ class CoreRepository(PulpBase): class CoreRepositoryVersion(PulpBase): __tablename__ = "core_repositoryversion" - pulp_id = sqlalchemy.Column( + pulp_id: Mapped[uuid.UUID] = mapped_column( UUID(as_uuid=True), primary_key=True, default=uuid.uuid4, ) - pulp_created = sqlalchemy.Column(sqlalchemy.DATETIME, default=datetime.now) - pulp_last_updated = sqlalchemy.Column( + pulp_created: Mapped[datetime.datetime] = mapped_column( + sqlalchemy.DATETIME, default=datetime.datetime.now + ) + pulp_last_updated: Mapped[datetime.datetime] = mapped_column( sqlalchemy.DATETIME, - default=datetime.now, + default=datetime.datetime.now, + ) + number: Mapped[int] = mapped_column(sqlalchemy.Integer) + complete: Mapped[bool] = mapped_column(sqlalchemy.Boolean) + base_version_id: Mapped[Optional[uuid.UUID]] = mapped_column( + UUID(as_uuid=True), nullable=True ) - number = sqlalchemy.Column(sqlalchemy.Integer) - complete = sqlalchemy.Column(sqlalchemy.Boolean) - base_version_id = sqlalchemy.Column(UUID(as_uuid=True), nullable=True) - repository_id = sqlalchemy.Column( + repository_id: Mapped[uuid.UUID] = mapped_column( UUID(as_uuid=True), sqlalchemy.ForeignKey(CoreRepository.pulp_id), ) - info = sqlalchemy.Column(JSONB) + info: Mapped[Dict[str, Any]] = mapped_column(JSONB) - repository = relationship( + repository: Mapped[CoreRepository] = relationship( CoreRepository, back_populates="versions", ) @@ -170,34 +210,36 @@ class CoreRepositoryVersion(PulpBase): class CoreContent(PulpBase): __tablename__ = "core_content" - pulp_id = sqlalchemy.Column( + pulp_id: Mapped[uuid.UUID] = mapped_column( UUID(as_uuid=True), primary_key=True, default=uuid.uuid4, ) - pulp_created = sqlalchemy.Column(sqlalchemy.DATETIME, default=datetime.now) - pulp_last_updated = sqlalchemy.Column( + pulp_created: Mapped[datetime.datetime] = mapped_column( + sqlalchemy.DATETIME, default=datetime.datetime.now + ) + pulp_last_updated: Mapped[datetime.datetime] = mapped_column( sqlalchemy.DATETIME, - default=datetime.now, + default=datetime.datetime.now, ) - pulp_type = sqlalchemy.Column(sqlalchemy.Text) - upstream_id = sqlalchemy.Column( + pulp_type: Mapped[str] = mapped_column(sqlalchemy.Text) + upstream_id: Mapped[Optional[uuid.UUID]] = mapped_column( UUID(as_uuid=True), nullable=True, ) - timestamp_of_interest = sqlalchemy.Column( + timestamp_of_interest: Mapped[datetime.datetime] = mapped_column( sqlalchemy.DATETIME, - default=datetime.now, + default=datetime.datetime.now, ) - rpm_package: "RpmPackage" = relationship( + rpm_package: Mapped["RpmPackage"] = relationship( "RpmPackage", back_populates="content", ) - core_contentartifact: "CoreContentArtifact" = relationship( + core_contentartifact: Mapped[List["CoreContentArtifact"]] = relationship( "CoreContentArtifact", back_populates="content", ) - core_repositorycontent: "CoreRepositoryContent" = relationship( + core_repositorycontent: Mapped["CoreRepositoryContent"] = relationship( "CoreRepositoryContent", back_populates="content", ) @@ -210,33 +252,35 @@ def file_href(self): class CoreContentArtifact(PulpBase): __tablename__ = "core_contentartifact" - pulp_id = sqlalchemy.Column( + pulp_id: Mapped[uuid.UUID] = mapped_column( UUID(as_uuid=True), primary_key=True, default=uuid.uuid4, ) - pulp_created = sqlalchemy.Column(sqlalchemy.DATETIME, default=datetime.now) - pulp_last_updated = sqlalchemy.Column( + pulp_created: Mapped[datetime.datetime] = mapped_column( + sqlalchemy.DATETIME, default=datetime.datetime.now + ) + pulp_last_updated: Mapped[Optional[datetime.datetime]] = mapped_column( sqlalchemy.DATETIME, - default=datetime.now, + default=datetime.datetime.now, nullable=True, ) - relative_path = sqlalchemy.Column(sqlalchemy.Text) - artifact_id = sqlalchemy.Column( + relative_path: Mapped[Optional[uuid.UUID]] = mapped_column(sqlalchemy.Text) + artifact_id: Mapped[Optional[uuid.UUID]] = mapped_column( UUID(as_uuid=True), sqlalchemy.ForeignKey("core_artifact.pulp_id"), nullable=True, ) - content_id = sqlalchemy.Column( + content_id: Mapped[uuid.UUID] = mapped_column( UUID(as_uuid=True), sqlalchemy.ForeignKey(CoreContent.pulp_id), ) - artifact: "CoreArtifact" = relationship( + artifact: Mapped["CoreArtifact"] = relationship( "CoreArtifact", foreign_keys=[artifact_id], back_populates="core_contentartifact", ) - content: CoreContent = relationship( + content: Mapped[CoreContent] = relationship( CoreContent, foreign_keys=[content_id], back_populates="core_contentartifact", @@ -246,30 +290,42 @@ class CoreContentArtifact(PulpBase): class CoreArtifact(PulpBase): __tablename__ = "core_artifact" - pulp_id = sqlalchemy.Column( + pulp_id: Mapped[uuid.UUID] = mapped_column( UUID(as_uuid=True), primary_key=True, default=uuid.uuid4, ) - pulp_created = sqlalchemy.Column(sqlalchemy.DATETIME, default=datetime.now) - pulp_last_updated = sqlalchemy.Column( + pulp_created: Mapped[datetime.datetime] = mapped_column( + sqlalchemy.DATETIME, default=datetime.datetime.now + ) + pulp_last_updated: Mapped[Optional[datetime.datetime]] = mapped_column( sqlalchemy.DATETIME, - default=datetime.now, + default=datetime.datetime.now, nullable=True, ) - file = sqlalchemy.Column(sqlalchemy.VARCHAR(255)) - size = sqlalchemy.Column(sqlalchemy.BigInteger) - md5 = sqlalchemy.Column(sqlalchemy.VARCHAR(32), nullable=True) - sha1 = sqlalchemy.Column(sqlalchemy.VARCHAR(40), nullable=True) - sha224 = sqlalchemy.Column(sqlalchemy.VARCHAR(56), nullable=True) - sha256 = sqlalchemy.Column(sqlalchemy.VARCHAR(64)) - sha384 = sqlalchemy.Column(sqlalchemy.VARCHAR(96), nullable=True) - sha512 = sqlalchemy.Column(sqlalchemy.VARCHAR(128), nullable=True) - timestamp_of_interest = sqlalchemy.Column( + file: Mapped[str] = mapped_column(sqlalchemy.VARCHAR(255)) + size: Mapped[int] = mapped_column(sqlalchemy.BigInteger) + md5: Mapped[Optional[str]] = mapped_column( + sqlalchemy.VARCHAR(32), nullable=True + ) + sha1: Mapped[Optional[str]] = mapped_column( + sqlalchemy.VARCHAR(40), nullable=True + ) + sha224: Mapped[Optional[str]] = mapped_column( + sqlalchemy.VARCHAR(56), nullable=True + ) + sha256: Mapped[str] = mapped_column(sqlalchemy.VARCHAR(64)) + sha384: Mapped[Optional[str]] = mapped_column( + sqlalchemy.VARCHAR(96), nullable=True + ) + sha512: Mapped[Optional[str]] = mapped_column( + sqlalchemy.VARCHAR(128), nullable=True + ) + timestamp_of_interest: Mapped[datetime.datetime] = mapped_column( sqlalchemy.DATETIME, - default=datetime.now, + default=datetime.datetime.now, ) - core_contentartifact: CoreContentArtifact = relationship( + core_contentartifact: Mapped[List[CoreContentArtifact]] = relationship( CoreContentArtifact, back_populates="artifact", ) @@ -278,49 +334,51 @@ class CoreArtifact(PulpBase): class CoreRepositoryContent(PulpBase): __tablename__ = "core_repositorycontent" - pulp_id = sqlalchemy.Column( + pulp_id: Mapped[uuid.UUID] = mapped_column( UUID(as_uuid=True), primary_key=True, default=uuid.uuid4, ) - pulp_created = sqlalchemy.Column(sqlalchemy.DATETIME, default=datetime.now) - pulp_last_updated = sqlalchemy.Column( + pulp_created: Mapped[datetime.datetime] = mapped_column( + sqlalchemy.DATETIME, default=datetime.datetime.now + ) + pulp_last_updated: Mapped[datetime.datetime] = mapped_column( sqlalchemy.DATETIME, - default=datetime.now, + default=datetime.datetime.now, ) - content_id = sqlalchemy.Column( + content_id: Mapped[uuid.UUID] = mapped_column( UUID(as_uuid=True), sqlalchemy.ForeignKey(CoreContent.pulp_id), ) - repository_id = sqlalchemy.Column( + repository_id: Mapped[uuid.UUID] = mapped_column( UUID(as_uuid=True), sqlalchemy.ForeignKey(CoreRepository.pulp_id), ) - version_added_id = sqlalchemy.Column( + version_added_id: Mapped[uuid.UUID] = mapped_column( UUID(as_uuid=True), sqlalchemy.ForeignKey(CoreRepositoryVersion.pulp_id), ) - version_removed_id = sqlalchemy.Column( + version_removed_id: Mapped[Optional[uuid.UUID]] = mapped_column( UUID(as_uuid=True), sqlalchemy.ForeignKey(CoreRepositoryVersion.pulp_id), nullable=True, ) - content: List[CoreContent] = relationship( + content: Mapped[List[CoreContent]] = relationship( CoreContent, foreign_keys=[content_id], back_populates="core_repositorycontent", ) - repository: List[CoreRepository] = relationship( + repository: Mapped[List[CoreRepository]] = relationship( CoreRepository, foreign_keys=[repository_id], back_populates="repository_content", ) - added_version: List[CoreRepositoryVersion] = relationship( + added_version: Mapped[List[CoreRepositoryVersion]] = relationship( CoreRepositoryVersion, foreign_keys=[version_added_id], ) - removed_version: List[CoreRepositoryVersion] = relationship( + removed_version: Mapped[List[CoreRepositoryVersion]] = relationship( CoreRepositoryVersion, foreign_keys=[version_removed_id], ) @@ -329,49 +387,63 @@ class CoreRepositoryContent(PulpBase): class RpmPackage(PulpBase): __tablename__ = "rpm_package" - content_ptr_id = sqlalchemy.Column( + content_ptr_id = mapped_column( UUID(as_uuid=True), sqlalchemy.ForeignKey(CoreContent.pulp_id), primary_key=True, ) - name = sqlalchemy.Column(sqlalchemy.VARCHAR(255)) - epoch = sqlalchemy.Column(sqlalchemy.VARCHAR(10)) - version = sqlalchemy.Column(sqlalchemy.VARCHAR(255)) - release = sqlalchemy.Column(sqlalchemy.VARCHAR(255)) - arch = sqlalchemy.Column(sqlalchemy.VARCHAR(20)) - pkgId = sqlalchemy.Column(sqlalchemy.Text) - checksum_type = sqlalchemy.Column(sqlalchemy.Text) - summary = sqlalchemy.Column(sqlalchemy.Text) - description = sqlalchemy.Column(sqlalchemy.Text) - url = sqlalchemy.Column(sqlalchemy.Text) - # changelogs = sqlalchemy.Column(JSONB) - # files = sqlalchemy.Column(JSONB) - # requires = sqlalchemy.Column(JSONB) - # provides = sqlalchemy.Column(JSONB) - # conflicts = sqlalchemy.Column(JSONB) - # obsoletes = sqlalchemy.Column(JSONB) - # suggests = sqlalchemy.Column(JSONB) - # enhances = sqlalchemy.Column(JSONB) - # recommends = sqlalchemy.Column(JSONB) - # supplements = sqlalchemy.Column(JSONB) - location_base = sqlalchemy.Column(sqlalchemy.Text) - location_href = sqlalchemy.Column(sqlalchemy.Text) - rpm_buildhost = sqlalchemy.Column(sqlalchemy.Text) - rpm_group = sqlalchemy.Column(sqlalchemy.Text) - rpm_license = sqlalchemy.Column(sqlalchemy.Text) - rpm_packager = sqlalchemy.Column(sqlalchemy.Text) - rpm_sourcerpm = sqlalchemy.Column(sqlalchemy.Text) - rpm_vendor = sqlalchemy.Column(sqlalchemy.Text) - rpm_header_start = sqlalchemy.Column(sqlalchemy.BigInteger, nullable=True) - rpm_header_end = sqlalchemy.Column(sqlalchemy.BigInteger, nullable=True) - is_modular = sqlalchemy.Column(sqlalchemy.Boolean) - size_archive = sqlalchemy.Column(sqlalchemy.BigInteger, nullable=True) - size_installed = sqlalchemy.Column(sqlalchemy.BigInteger, nullable=True) - size_package = sqlalchemy.Column(sqlalchemy.BigInteger, nullable=True) - time_build = sqlalchemy.Column(sqlalchemy.BigInteger, nullable=True) - time_file = sqlalchemy.Column(sqlalchemy.BigInteger, nullable=True) - - content: CoreContent = relationship( + name: Mapped[str] = mapped_column(sqlalchemy.VARCHAR(255)) + epoch: Mapped[str] = mapped_column(sqlalchemy.VARCHAR(10)) + version: Mapped[str] = mapped_column(sqlalchemy.VARCHAR(255)) + release: Mapped[str] = mapped_column(sqlalchemy.VARCHAR(255)) + arch: Mapped[str] = mapped_column(sqlalchemy.VARCHAR(20)) + pkgId: Mapped[str] = mapped_column(sqlalchemy.Text) + checksum_type: Mapped[str] = mapped_column(sqlalchemy.Text) + summary: Mapped[str] = mapped_column(sqlalchemy.Text) + description: Mapped[str] = mapped_column(sqlalchemy.Text) + url: Mapped[str] = mapped_column(sqlalchemy.Text) + # changelogs = mapped_column(JSONB) + # files = mapped_column(JSONB) + # requires = mapped_column(JSONB) + # provides = mapped_column(JSONB) + # conflicts = mapped_column(JSONB) + # obsoletes = mapped_column(JSONB) + # suggests = mapped_column(JSONB) + # enhances = mapped_column(JSONB) + # recommends = mapped_column(JSONB) + # supplements = mapped_column(JSONB) + location_base: Mapped[str] = mapped_column(sqlalchemy.Text) + location_href: Mapped[str] = mapped_column(sqlalchemy.Text) + rpm_buildhost: Mapped[str] = mapped_column(sqlalchemy.Text) + rpm_group: Mapped[str] = mapped_column(sqlalchemy.Text) + rpm_license: Mapped[str] = mapped_column(sqlalchemy.Text) + rpm_packager: Mapped[str] = mapped_column(sqlalchemy.Text) + rpm_sourcerpm: Mapped[str] = mapped_column(sqlalchemy.Text) + rpm_vendor: Mapped[str] = mapped_column(sqlalchemy.Text) + rpm_header_start: Mapped[Optional[int]] = mapped_column( + sqlalchemy.BigInteger, nullable=True + ) + rpm_header_end: Mapped[Optional[int]] = mapped_column( + sqlalchemy.BigInteger, nullable=True + ) + is_modular: Mapped[bool] = mapped_column(sqlalchemy.Boolean) + size_archive: Mapped[Optional[int]] = mapped_column( + sqlalchemy.BigInteger, nullable=True + ) + size_installed: Mapped[Optional[int]] = mapped_column( + sqlalchemy.BigInteger, nullable=True + ) + size_package: Mapped[Optional[int]] = mapped_column( + sqlalchemy.BigInteger, nullable=True + ) + time_build: Mapped[Optional[int]] = mapped_column( + sqlalchemy.BigInteger, nullable=True + ) + time_file: Mapped[Optional[int]] = mapped_column( + sqlalchemy.BigInteger, nullable=True + ) + + content: Mapped[CoreContent] = relationship( CoreContent, back_populates="rpm_package", ) @@ -412,16 +484,16 @@ def as_dict(self) -> Dict[str, Any]: class RpmModulemd(PulpBase): __tablename__ = "rpm_modulemd" - content_ptr_id = sqlalchemy.Column( + content_ptr_id: Mapped[uuid.UUID] = mapped_column( UUID(as_uuid=True), sqlalchemy.ForeignKey(CoreContent.pulp_id), primary_key=True, ) - name = sqlalchemy.Column(sqlalchemy.Text) - stream = sqlalchemy.Column(sqlalchemy.Text) - version = sqlalchemy.Column(sqlalchemy.Text) - context = sqlalchemy.Column(sqlalchemy.Text) - arch = sqlalchemy.Column(sqlalchemy.Text) + name: Mapped[str] = mapped_column(sqlalchemy.Text) + stream: Mapped[str] = mapped_column(sqlalchemy.Text) + version: Mapped[str] = mapped_column(sqlalchemy.Text) + context: Mapped[str] = mapped_column(sqlalchemy.Text) + arch: Mapped[str] = mapped_column(sqlalchemy.Text) @property def nsvca(self): @@ -431,12 +503,12 @@ def nsvca(self): class RpmModulemdPackages(PulpBase): __tablename__ = "rpm_modulemd_packages" - id = sqlalchemy.Column(sqlalchemy.Integer, primary_key=True) - modulemd_id = sqlalchemy.Column( + id: Mapped[int] = mapped_column(sqlalchemy.Integer, primary_key=True) + modulemd_id: Mapped[uuid.UUID] = mapped_column( UUID(as_uuid=True), sqlalchemy.ForeignKey(RpmModulemd.content_ptr_id), ) - package_id = sqlalchemy.Column( + package_id: Mapped[uuid.UUID] = mapped_column( UUID(as_uuid=True), sqlalchemy.ForeignKey(RpmPackage.content_ptr_id), ) diff --git a/requirements.txt b/requirements.txt index 7df2dd2f5..cf478cd7c 100644 --- a/requirements.txt +++ b/requirements.txt @@ -1,13 +1,13 @@ fastapi==0.110.0 fastapi-users[all]==12.1.3 # Can be updated only after migration to SQLAlchemy > 2.0 -fastapi-users-db-sqlalchemy==4.0.5 +fastapi-users-db-sqlalchemy==6.0.1 uvicorn==0.27.1 asyncpg==0.29.0 psycopg2-binary==2.9.9 pydantic==2.6.3 pydantic-settings==2.2.1 -SQLAlchemy==1.4.51 +SQLAlchemy==2.0.28 aiohttp==3.9.3 aiohttp-retry==2.8.3 aiofiles==23.2.1