Coverage for daklib/dbconn.py: 87%
1555 statements
« prev ^ index » next coverage.py v7.6.0, created at 2026-08-03 16:46 +0000
1"""DB access class
3@contact: Debian FTPMaster <ftpmaster@debian.org>
4@copyright: 2000, 2001, 2002, 2003, 2004, 2006 James Troup <james@nocrew.org>
5@copyright: 2008-2009 Mark Hymers <mhy@debian.org>
6@copyright: 2009, 2010 Joerg Jaspert <joerg@debian.org>
7@copyright: 2009 Mike O'Connor <stew@debian.org>
8@license: GNU General Public License version 2 or later
9"""
11# This program is free software; you can redistribute it and/or modify
12# it under the terms of the GNU General Public License as published by
13# the Free Software Foundation; either version 2 of the License, or
14# (at your option) any later version.
16# This program is distributed in the hope that it will be useful,
17# but WITHOUT ANY WARRANTY; without even the implied warranty of
18# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
19# GNU General Public License for more details.
21# You should have received a copy of the GNU General Public License
22# along with this program; if not, write to the Free Software
23# Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA
25################################################################################
27# < mhy> I need a funny comment
28# < sgran> two peanuts were walking down a dark street
29# < sgran> one was a-salted
30# * mhy looks up the definition of "funny"
32################################################################################
34import functools
35import inspect
36import os
37import re
38import subprocess
39import warnings
40from collections.abc import Callable, Iterable
41from datetime import datetime, timedelta
42from os.path import normpath
43from tarfile import TarFile
44from typing import TYPE_CHECKING, Any, TypedDict, overload, override
46import apt_pkg
47import sqlalchemy
48import sqlalchemy.types
49from debian.deb822 import Deb822
50from sqlalchemy import (
51 Column,
52 ForeignKey,
53 Select,
54 Table,
55 UniqueConstraint,
56 create_engine,
57 desc,
58 exists,
59 select,
60 sql,
61)
62from sqlalchemy.dialects.postgresql import ARRAY
64# Don't remove this, we re-export the exceptions to scripts which import us
65from sqlalchemy.exc import IntegrityError, OperationalError, SAWarning, SQLAlchemyError
66from sqlalchemy.ext.associationproxy import AssociationProxy, association_proxy
67from sqlalchemy.orm import (
68 DynamicMapped,
69 Mapped,
70 attribute_keyed_dict,
71 mapped_column,
72 object_session,
73 relationship,
74 sessionmaker,
75 with_parent,
76)
77from sqlalchemy.sql import func
78from sqlalchemy.types import CHAR, BigInteger, Boolean, DateTime, Integer, String, Text
80import daklib.gpg
82from .aptversion import AptVersion
84# Only import Config until Queue stuff is changed to store its config
85# in the database
86from .config import Config
87from .textutils import fix_maintainer
89# suppress some deprecation warnings in squeeze related to sqlalchemy
90warnings.filterwarnings(
91 "ignore", "Predicate of partial index .* ignored during reflection", SAWarning
92)
94# (Debian 12 "bookworm") Silence warning targeted at SQLAlchemy dialect maintainers
95warnings.filterwarnings(
96 "ignore",
97 "Dialect postgresql:psycopg2 will not make use of SQL compilation caching.*",
98 SAWarning,
99)
101from .database.base import Base, BaseTimestamp
103if TYPE_CHECKING:
104 import sqlalchemy.orm.query
105 from sqlalchemy.orm import Session
108################################################################################
111class DebVersion(sqlalchemy.types.UserDefinedType):
112 @override
113 def get_col_spec(self, **kw: Any) -> str:
114 return "DEBVERSION"
116 @override
117 def bind_processor(self, dialect):
118 return None
120 @override
121 def result_processor(self, dialect, coltype):
122 return None
125################################################################################
127__all__ = ["IntegrityError", "SQLAlchemyError", "DebVersion"]
129################################################################################
132def session_wrapper[**P, R](fn: Callable[P, R]) -> Callable[P, R]:
133 """
134 Wrapper around common ".., session=None):" handling. If the wrapped
135 function is called without passing 'session', we create a local one
136 and destroy it when the function ends.
138 Also attaches a commit_or_flush method to the session; if we created a
139 local session, this is a synonym for session.commit(), otherwise it is a
140 synonym for session.flush().
141 """
143 @functools.wraps(fn)
144 def wrapped(*args, **kwargs):
145 private_transaction = False
147 # Find the session object
148 session = kwargs.get("session")
150 if session is None:
151 if len(args) < len(inspect.getfullargspec(fn).args):
152 # No session specified as last argument or in kwargs
153 private_transaction = True
154 kwargs["session"] = session = DBConn().session()
155 else:
156 # Session is last argument in args
157 session = args[-1]
158 if session is None: 158 ↛ 159line 158 didn't jump to line 159 because the condition on line 158 was never true
159 session = DBConn().session()
160 args = (*args[:-1], session)
161 private_transaction = True
163 if private_transaction:
164 session.commit_or_flush = session.commit # type: ignore[union-attr]
165 else:
166 session.commit_or_flush = session.flush # type: ignore[union-attr]
168 try:
169 return fn(*args, **kwargs)
170 finally:
171 if private_transaction:
172 # We created a session; close it.
173 session.close()
175 return wrapped
178__all__.append("session_wrapper")
180################################################################################
183class ORMObject(Base):
184 """
185 ORMObject is a base class for all ORM classes mapped by SQLalchemy. All
186 derived classes must implement the properties() method.
187 """
189 __abstract__ = True
191 def properties(self) -> list[str]:
192 """
193 This method should be implemented by all derived classes and returns a
194 list of the important properties. The properties 'created' and
195 'modified' will be added automatically. A suffix '_count' should be
196 added to properties that are lists or query objects. The most important
197 property name should be returned as the first element in the list
198 because it is used by repr().
199 """
200 return []
202 def classname(self) -> str:
203 """
204 Returns the name of the class.
205 """
206 return type(self).__name__
208 @override
209 def __repr__(self):
210 """
211 Returns a short string representation of the object using the first
212 element from the properties() method.
213 """
214 primary_property = self.properties()[0]
215 value = getattr(self, primary_property)
216 return "<%s %s>" % (self.classname(), str(value))
218 @override
219 def __str__(self):
220 """
221 Returns a human readable form of the object using the properties()
222 method.
223 """
224 return "<%s(...)>" % (self.classname())
226 @classmethod
227 @session_wrapper
228 def get(cls, primary_key, session: "Session | None" = None):
229 """
230 This is a support function that allows getting an object by its primary
231 key.
233 Architecture.get(3[, session])
235 instead of the more verbose
237 session.get(Architecture, 3)
238 """
239 assert session is not None
240 return session.get(cls, primary_key)
242 def session(self) -> "Session | None":
243 """
244 Returns the current session that is associated with the object. May
245 return None is object is in detached state.
246 """
248 return object_session(self)
251__all__.append("ORMObject")
253################################################################################
256class ACL(ORMObject):
257 __tablename__ = "acl"
259 id: Mapped[int] = mapped_column(primary_key=True)
260 name: Mapped[str] = mapped_column() # TODO: add unique=True
261 is_global: Mapped[bool] = mapped_column(default=False)
262 match_fingerprint: Mapped[bool] = mapped_column(default=False)
263 match_keyring_id: Mapped[int | None] = mapped_column(ForeignKey("keyrings.id"))
264 allow_new: Mapped[bool] = mapped_column(default=False)
265 allow_source: Mapped[bool] = mapped_column(default=False)
266 allow_binary: Mapped[bool] = mapped_column(default=False)
267 allow_binary_all: Mapped[bool] = mapped_column(default=False)
268 allow_binary_only: Mapped[bool] = mapped_column(default=False)
269 allow_hijack: Mapped[bool] = mapped_column(default=False)
270 allow_per_source: Mapped[bool] = mapped_column(default=False)
271 deny_per_source: Mapped[bool] = mapped_column(default=False)
273 architectures: Mapped[set["Architecture"]] = relationship(
274 secondary="acl_architecture_map"
275 )
276 fingerprints: Mapped[set["Fingerprint"]] = relationship(
277 secondary="acl_fingerprint_map"
278 )
279 match_keyring: Mapped["Keyring"] = relationship(foreign_keys=[match_keyring_id])
280 per_source: Mapped[set["ACLPerSource"]] = relationship(back_populates="acl")
281 per_suite: Mapped[set["ACLPerSuite"]] = relationship(back_populates="acl")
283 @override
284 def __repr__(self):
285 return "<ACL {0}>".format(self.name)
288__all__.append("ACL")
291class AclArchitectureMap(Base):
292 __tablename__ = "acl_architecture_map"
294 acl_id: Mapped[int] = mapped_column(
295 ForeignKey("acl.id", ondelete="CASCADE"), primary_key=True
296 )
297 architecture_id: Mapped[int] = mapped_column(
298 ForeignKey("architecture.id", ondelete="CASCADE"), primary_key=True
299 )
302class AclFingerprintMap(Base):
303 __tablename__ = "acl_fingerprint_map"
305 acl_id: Mapped[int] = mapped_column(
306 ForeignKey("acl.id", ondelete="CASCADE"), primary_key=True
307 )
308 fingerprint_id: Mapped[int] = mapped_column(
309 ForeignKey("fingerprint.id", ondelete="CASCADE"), primary_key=True
310 )
313class ACLPerSource(ORMObject):
314 __tablename__ = "acl_per_source"
316 acl_id: Mapped[int] = mapped_column(
317 ForeignKey("acl.id", ondelete="CASCADE"), primary_key=True
318 )
319 fingerprint_id: Mapped[int] = mapped_column(
320 ForeignKey("fingerprint.id", ondelete="CASCADE"), primary_key=True
321 )
322 source: Mapped[str] = mapped_column(primary_key=True)
323 reason: Mapped[str | None] = mapped_column(default=None)
324 created_by_id: Mapped[int | None] = mapped_column(ForeignKey("fingerprint.id"))
325 created: Mapped[datetime] = mapped_column(
326 DateTime(timezone=False), server_default=func.now()
327 )
329 acl: Mapped[ACL] = relationship(back_populates="per_source")
330 fingerprint: Mapped["Fingerprint"] = relationship(foreign_keys=[fingerprint_id])
331 created_by: Mapped["Fingerprint"] = relationship(foreign_keys=[created_by_id])
333 @override
334 def __repr__(self):
335 return "<ACLPerSource acl={0} fingerprint={1} source={2} reason={3}>".format(
336 self.acl.name, self.fingerprint.fingerprint, self.source, self.reason
337 )
340__all__.append("ACLPerSource")
343class ACLPerSuite(ORMObject):
344 __tablename__ = "acl_per_suite"
346 acl_id: Mapped[int] = mapped_column(
347 ForeignKey("acl.id", ondelete="CASCADE"), primary_key=True
348 )
349 fingerprint_id: Mapped[int] = mapped_column(
350 ForeignKey("fingerprint.id", ondelete="CASCADE"), primary_key=True
351 )
352 suite_id: Mapped[int] = mapped_column(
353 ForeignKey("suite.id", ondelete="CASCADE"), primary_key=True
354 )
355 reason: Mapped[str | None] = mapped_column(default=None)
356 created_by_id: Mapped[int | None] = mapped_column(ForeignKey("fingerprint.id"))
357 created: Mapped[datetime] = mapped_column(
358 DateTime(timezone=False), server_default=func.now()
359 )
361 acl: Mapped[ACL] = relationship(back_populates="per_suite")
362 fingerprint: Mapped["Fingerprint"] = relationship(foreign_keys=[fingerprint_id])
363 suite: Mapped["Suite"] = relationship()
364 created_by: Mapped["Fingerprint"] = relationship(foreign_keys=[created_by_id])
366 @override
367 def __repr__(self):
368 return "<ACLPerSuite acl={0} fingerprint={1} suite={2} reason={3}>".format(
369 self.acl.name,
370 self.fingerprint.fingerprint,
371 self.suite.suite_name,
372 self.reason,
373 )
376__all__.append("ACLPerSuite")
378################################################################################
381class Architecture(BaseTimestamp):
382 __tablename__ = "architecture"
384 arch_id: Mapped[int] = mapped_column("id", Integer(), primary_key=True)
385 arch_string: Mapped[str] = mapped_column(Text(), unique=True)
386 description: Mapped[str]
388 suites: Mapped[list["Suite"]] = relationship(
389 secondary="suite_architectures", back_populates="architectures"
390 )
392 def __init__(self, arch_string=None, description=None):
393 self.arch_string = arch_string
394 self.description = description
396 @override
397 def __str__(self):
398 return self.arch_string
400 @override
401 def __repr__(self):
402 return "<{} {}>".format(
403 self.__class__.__name__,
404 self.arch_string,
405 )
407 @override
408 def __eq__(self, val):
409 if isinstance(val, str):
410 warnings.warn(
411 "comparison with a `str` is deprecated",
412 DeprecationWarning,
413 stacklevel=2,
414 )
415 return self.arch_string == val
416 # This signals to use the normal comparison operator
417 return NotImplemented
419 @override
420 def __ne__(self, val):
421 if isinstance(val, str): 421 ↛ 429line 421 didn't jump to line 429 because the condition on line 421 was always true
422 warnings.warn(
423 "comparison with a `str` is deprecated",
424 DeprecationWarning,
425 stacklevel=2,
426 )
427 return self.arch_string != val
428 # This signals to use the normal comparison operator
429 return NotImplemented
431 __hash__ = BaseTimestamp.__hash__
434__all__.append("Architecture")
437@session_wrapper
438def get_architecture(
439 architecture: str, session: "Session | None" = None
440) -> Architecture | None:
441 """
442 Returns database id for given `architecture`.
444 :param architecture: The name of the architecture
445 :param session: Optional SQLA session object (a temporary one will be
446 generated if not supplied)
447 :return: Architecture object for the given arch (None if not present)
448 """
450 assert session is not None
451 q = select(Architecture).where(Architecture.arch_string == architecture)
452 return session.execute(q).scalar_one_or_none()
455__all__.append("get_architecture")
457################################################################################
460class Archive(Base):
461 __tablename__ = "archive"
463 archive_id: Mapped[int] = mapped_column("id", primary_key=True)
464 archive_name: Mapped[str] = mapped_column("name", unique=True)
465 description: Mapped[str | None] = mapped_column(default=None)
466 created: Mapped[datetime] = mapped_column(
467 DateTime(timezone=True), server_default=func.now()
468 )
469 modified: Mapped[datetime] = mapped_column(
470 DateTime(timezone=True), server_default=func.now()
471 )
472 primary_mirror: Mapped[str | None] = mapped_column(default=None)
473 path: Mapped[str]
474 mode: Mapped[str] = mapped_column(CHAR(length=4), default="0644")
475 tainted: Mapped[bool] = mapped_column(default=False)
476 use_morgue: Mapped[bool] = mapped_column(default=True)
477 stayofexecution: Mapped[timedelta] = mapped_column(default=timedelta(hours=36))
478 changelog: Mapped[str | None]
480 files: Mapped[list["ArchiveFile"]] = relationship(back_populates="archive")
481 suites: Mapped[list["Suite"]] = relationship(back_populates="archive")
483 def __init__(self, *args, **kwargs):
484 pass
486 @override
487 def __repr__(self):
488 return "<Archive %s>" % self.archive_name
491__all__.append("Archive")
494@session_wrapper
495def get_archive(archive: str, session: "Session | None" = None) -> Archive | None:
496 """
497 returns database id for given `archive`.
499 :param archive: the name of the arhive
500 :param session: Optional SQLA session object (a temporary one will be
501 generated if not supplied)
502 :return: Archive object for the given name (None if not present)
503 """
505 assert session is not None
506 archive = archive.lower()
507 q = select(Archive).where(Archive.archive_name == archive)
508 return session.execute(q).scalar_one_or_none()
511__all__.append("get_archive")
513################################################################################
516class ArchiveFile(Base):
517 __tablename__ = "files_archive_map"
519 file_id: Mapped[int] = mapped_column(ForeignKey("files.id"), primary_key=True)
520 archive_id: Mapped[int] = mapped_column(ForeignKey("archive.id"), primary_key=True)
521 component_id: Mapped[int] = mapped_column(
522 ForeignKey("component.id"), primary_key=True
523 )
524 last_used: Mapped[datetime | None] = mapped_column(
525 DateTime(timezone=False), default=None
526 )
527 created: Mapped[datetime] = mapped_column(
528 DateTime(timezone=False), server_default=func.now()
529 )
531 archive: Mapped[Archive] = relationship(back_populates="files")
532 component: Mapped["Component"] = relationship()
533 file: Mapped["PoolFile"] = relationship(back_populates="archives")
535 def __init__(self, archive=None, component=None, file=None):
536 self.archive = archive
537 self.component = component
538 self.file = file
540 @property
541 def path(self):
542 return os.path.join(
543 self.archive.path, "pool", self.component.component_name, self.file.filename
544 )
547__all__.append("ArchiveFile")
549################################################################################
552class BinContents(ORMObject):
553 __tablename__ = "bin_contents"
555 binary_id: Mapped[int] = mapped_column(
556 ForeignKey("binaries.id", ondelete="CASCADE"), primary_key=True
557 )
558 file: Mapped[str] = mapped_column(primary_key=True)
560 binary: Mapped["DBBinary"] = relationship(back_populates="contents")
562 def __init__(self, file=None, binary=None):
563 self.file = file
564 self.binary = binary
566 @override
567 def properties(self) -> list[str]:
568 return ["file", "binary"]
571__all__.append("BinContents")
573################################################################################
576class DBBinary(ORMObject):
577 __tablename__ = "binaries"
578 __table_args = (
579 UniqueConstraint(
580 "package", "version", "architecture", name="binaries_package_key"
581 ),
582 )
584 binary_id: Mapped[int] = mapped_column("id", primary_key=True)
585 package: Mapped[str]
586 version: Mapped[str] = mapped_column(DebVersion())
587 maintainer_id: Mapped[int] = mapped_column(
588 "maintainer", ForeignKey("maintainer.id")
589 )
590 source_id: Mapped[int] = mapped_column("source", ForeignKey("source.id"))
591 arch_id: Mapped[int] = mapped_column("architecture", ForeignKey("architecture.id"))
592 poolfile_id: Mapped[int] = mapped_column(
593 "file", ForeignKey("files.id"), unique=True
594 )
595 binarytype: Mapped[str] = mapped_column("type")
596 fingerprint_id: Mapped[int | None] = mapped_column(
597 "sig_fpr", ForeignKey("fingerprint.id")
598 )
599 install_date: Mapped[datetime | None] = mapped_column(
600 DateTime(timezone=True), server_default=func.now()
601 )
602 created: Mapped[datetime | None] = mapped_column(
603 DateTime(timezone=True), server_default=func.now()
604 )
605 modified: Mapped[datetime | None] = mapped_column(
606 DateTime(timezone=True), server_default=func.now()
607 )
608 stanza: Mapped[str]
609 authorized_by_fingerprint_id: Mapped[int | None] = mapped_column(
610 ForeignKey("fingerprint.id")
611 )
613 maintainer: Mapped["Maintainer"] = relationship()
614 source: Mapped["DBSource"] = relationship(back_populates="binaries")
615 architecture: Mapped[Architecture] = relationship()
616 poolfile: Mapped["PoolFile"] = relationship()
617 fingerprint: Mapped["Fingerprint"] = relationship(foreign_keys=[fingerprint_id])
618 authorized_by_fingerprint: Mapped["Fingerprint"] = relationship(
619 foreign_keys=[authorized_by_fingerprint_id]
620 )
621 suites: Mapped[list["Suite"]] = relationship(
622 secondary="bin_associations", back_populates="binaries"
623 )
624 extra_sources: Mapped[list["DBSource"]] = relationship(
625 secondary="extra_src_references", back_populates="extra_binary_references"
626 )
627 key: Mapped[dict["MetadataKey", "BinaryMetadata"]] = relationship(
628 collection_class=attribute_keyed_dict("key"), cascade="all"
629 )
630 contents: DynamicMapped["BinContents"] = relationship(
631 cascade="all", back_populates="binary"
632 )
634 metadata_proxy: AssociationProxy[dict["MetadataKey", str]] = association_proxy(
635 "key", "value"
636 )
638 def __init__(
639 self,
640 package=None,
641 source=None,
642 version=None,
643 maintainer=None,
644 architecture=None,
645 poolfile=None,
646 binarytype="deb",
647 fingerprint=None,
648 ):
649 self.package = package
650 self.source = source
651 self.version = version
652 self.maintainer = maintainer
653 self.architecture = architecture
654 self.poolfile = poolfile
655 self.binarytype = binarytype
656 self.fingerprint = fingerprint
658 @property
659 def pkid(self) -> int:
660 return self.binary_id
662 @property
663 def name(self) -> str:
664 return self.package
666 @property
667 def arch_string(self) -> str:
668 return "%s" % self.architecture
670 @override
671 def properties(self) -> list[str]:
672 return [
673 "package",
674 "version",
675 "maintainer",
676 "source",
677 "architecture",
678 "poolfile",
679 "binarytype",
680 "fingerprint",
681 "install_date",
682 "suites_count",
683 "binary_id",
684 "contents_count",
685 "extra_sources",
686 ]
688 def scan_contents(self) -> Iterable[str]:
689 """
690 Yields the contents of the package. Only regular files are yielded and
691 the path names are normalized after converting them from either utf-8
692 or iso8859-1 encoding. It yields the string ' <EMPTY PACKAGE>' if the
693 package does not contain any regular file.
694 """
695 fullpath = self.poolfile.fullpath
696 dpkg_cmd = ("dpkg-deb", "--fsys-tarfile", fullpath)
697 dpkg = subprocess.Popen(dpkg_cmd, stdout=subprocess.PIPE)
698 tar = TarFile.open(fileobj=dpkg.stdout, mode="r|")
699 for member in tar.getmembers():
700 if not member.isdir():
701 name = normpath(member.name)
702 yield name
703 tar.close()
704 assert dpkg.stdout is not None
705 dpkg.stdout.close()
706 dpkg.wait()
708 def read_control(self) -> bytes:
709 """
710 Reads the control information from a binary.
712 :return: stanza text of the control section.
713 """
714 from . import utils
716 fullpath = self.poolfile.fullpath
717 return utils.deb_extract_control(fullpath)
719 def read_control_fields(self) -> apt_pkg.TagSection:
720 """
721 Reads the control information from a binary and return
722 as a dictionary.
724 :return: fields of the control section as a dictionary.
725 """
726 stanza = self.read_control()
727 return apt_pkg.TagSection(stanza)
729 @property
730 def proxy(self) -> "MetadataProxy":
731 session = object_session(self)
732 assert session is not None
733 query = select(BinaryMetadata).filter_by(binary=self)
734 return MetadataProxy(session, query)
737__all__.append("DBBinary")
740@session_wrapper
741def get_suites_binary_in(
742 package: str, session: "Session | None" = None
743) -> "list[Suite]":
744 """
745 Returns list of Suite objects which given `package` name is in
747 :param package: DBBinary package name to search for
748 :return: list of Suite objects for the given package
749 """
751 assert session is not None
752 q = select(Suite).where(Suite.binaries.any(DBBinary.package == package))
753 return list(session.scalars(q))
756__all__.append("get_suites_binary_in")
759@session_wrapper
760def get_component_by_package_suite(
761 package: str,
762 suite_list: list[str],
763 arch_list: list[str] | None = None,
764 session: "Session | None" = None,
765) -> str | None:
766 """
767 Returns the component name of the newest binary package in suite_list or
768 None if no package is found. The result can be optionally filtered by a list
769 of architecture names.
771 :param package: DBBinary package name to search for
772 :param suite_list: list of suite_name items
773 :param arch_list: optional list of arch_string items that defaults to []
774 :return: name of component or None
775 """
777 assert session is not None
778 q = (
779 select(DBBinary)
780 .filter_by(package=package)
781 .join(DBBinary.suites)
782 .where(Suite.suite_name.in_(suite_list))
783 )
784 if arch_list:
785 q = q.join(DBBinary.architecture).where(Architecture.arch_string.in_(arch_list))
786 q = q.order_by(desc(DBBinary.version)).limit(1)
787 binary = session.scalars(q).first()
788 if binary is None:
789 return None
790 else:
791 return binary.poolfile.component.component_name
794__all__.append("get_component_by_package_suite")
797class BinAssociations(Base):
798 __tablename__ = "bin_associations"
799 __table_args = (
800 UniqueConstraint("suite", "bin", name="bin_associations_suite_key"),
801 )
803 id: Mapped[int] = mapped_column(primary_key=True)
804 suite: Mapped[int] = mapped_column(ForeignKey("suite.id"))
805 bin: Mapped[int] = mapped_column(ForeignKey("binaries.id"))
806 created: Mapped[datetime] = mapped_column(
807 DateTime(timezone=True), server_default=func.now()
808 )
809 modified: Mapped[datetime] = mapped_column(
810 DateTime(timezone=True), server_default=func.now()
811 )
814class ExtraSrcReferences(Base):
815 __tablename__ = "extra_src_references"
817 bin_id: Mapped[int] = mapped_column(
818 ForeignKey("binaries.id", ondelete="CASCADE"), primary_key=True
819 )
820 src_id: Mapped[int] = mapped_column(ForeignKey("source.id"), primary_key=True)
823################################################################################
826class BuildQueue(Base):
827 __tablename__ = "build_queue"
829 queue_id: Mapped[int] = mapped_column("id", primary_key=True)
830 queue_name: Mapped[str] = mapped_column(unique=True)
831 generate_metadata: Mapped[bool] = mapped_column(default=False)
832 stay_of_execution: Mapped[int] = mapped_column(default=86400)
833 created: Mapped[datetime | None] = mapped_column(
834 DateTime(timezone=True), server_default=func.now()
835 )
836 modified: Mapped[datetime | None] = mapped_column(
837 DateTime(timezone=True), server_default=func.now()
838 )
839 suite_id: Mapped[int] = mapped_column(ForeignKey("suite.id"))
841 suite: Mapped["Suite"] = relationship()
843 def __init__(self, *args, **kwargs):
844 pass
846 @override
847 def __repr__(self):
848 return "<BuildQueue %s>" % self.queue_name
851__all__.append("BuildQueue")
853################################################################################
856class Component(ORMObject):
857 __tablename__ = "component"
859 component_id: Mapped[int] = mapped_column("id", primary_key=True)
860 component_name: Mapped[str] = mapped_column("name", unique=True)
861 description: Mapped[str | None]
862 meets_dfsg: Mapped[bool | None]
863 created: Mapped[datetime | None] = mapped_column(
864 DateTime(timezone=True), server_default=func.now()
865 )
866 modified: Mapped[datetime | None] = mapped_column(
867 DateTime(timezone=True), server_default=func.now()
868 )
869 ordering: Mapped[int] = mapped_column(unique=True)
871 suites: Mapped[list["Suite"]] = relationship(
872 secondary="component_suite", back_populates="components"
873 )
874 overrides: DynamicMapped["Override"] = relationship(back_populates="component")
876 def __init__(self, component_name=None):
877 self.component_name = component_name
879 @override
880 def __eq__(self, val: object):
881 if isinstance(val, str): 881 ↛ 882line 881 didn't jump to line 882 because the condition on line 881 was never true
882 warnings.warn(
883 "comparison with a `str` is deprecated",
884 DeprecationWarning,
885 stacklevel=2,
886 )
887 return self.component_name == val
888 # This signals to use the normal comparison operator
889 return NotImplemented
891 @override
892 def __ne__(self, val: object):
893 if isinstance(val, str):
894 warnings.warn(
895 "comparison with a `str` is deprecated",
896 DeprecationWarning,
897 stacklevel=2,
898 )
899 return self.component_name != val
900 # This signals to use the normal comparison operator
901 return NotImplemented
903 __hash__ = ORMObject.__hash__
905 @override
906 def properties(self) -> list[str]:
907 return [
908 "component_name",
909 "component_id",
910 "description",
911 "meets_dfsg",
912 "overrides_count",
913 ]
916__all__.append("Component")
919@session_wrapper
920def get_component(component: str, session: "Session | None" = None) -> Component | None:
921 """
922 Returns database id for given `component`.
924 :param component: The name of the override type
925 :return: the database id for the given component
926 """
928 assert session is not None
929 component = component.lower()
930 q = select(Component).where(Component.component_name == component)
931 return session.execute(q).scalar_one_or_none()
934__all__.append("get_component")
937@session_wrapper
938def get_component_by_name(
939 component_name: str, session: "Session | None" = None
940) -> Component:
941 """
942 Returns the Component with the given name.
944 Unlike get_component() the name is used as-is (no lower-casing) and an
945 exception is raised if the component does not exist.
947 :param component_name: the name of the component
948 :return: the Component object for the given name
949 """
951 assert session is not None
952 q = select(Component).where(Component.component_name == component_name)
953 return session.execute(q).scalar_one()
956__all__.append("get_component_by_name")
959def get_mapped_component_name(component_name: str) -> str:
960 cnf = Config()
961 for m in cnf.value_list("ComponentMappings"): 961 ↛ 962line 961 didn't jump to line 962 because the loop on line 961 never started
962 (src, dst) = m.split()
963 if component_name == src:
964 component_name = dst
965 return component_name
968__all__.append("get_mapped_component_name")
971@session_wrapper
972def get_mapped_component(
973 component_name: str, session: "Session | None" = None
974) -> Component | None:
975 """get component after mappings
977 Evaluate component mappings from ComponentMappings in dak.conf for the
978 given component name.
980 .. todo::
982 ansgar wants to get rid of this. It's currently only used for
983 the security archive
985 :param component_name: component name
986 :param session: database session
987 :return: component after applying maps or :const:`None`
988 """
989 assert session is not None
990 component_name = get_mapped_component_name(component_name)
991 q = select(Component).where(Component.component_name == component_name).limit(1)
992 return session.scalars(q).first()
995__all__.append("get_mapped_component")
998@session_wrapper
999def get_component_names(session: "Session | None" = None) -> list[str]:
1000 """
1001 Returns list of strings of component names.
1003 :return: list of strings of component names
1004 """
1006 assert session is not None
1007 return list(session.scalars(select(Component.component_name)))
1010__all__.append("get_component_names")
1013class ComponentSuite(Base):
1014 __tablename__ = "component_suite"
1016 component_id: Mapped[int] = mapped_column(
1017 ForeignKey("component.id", ondelete="CASCADE"), primary_key=True
1018 )
1019 suite_id: Mapped[int] = mapped_column(
1020 ForeignKey("suite.id", ondelete="CASCADE"), primary_key=True
1021 )
1024################################################################################
1027class DBConfig(Base):
1028 __tablename__ = "config"
1030 config_id: Mapped[int] = mapped_column("id", primary_key=True)
1031 name: Mapped[str] = mapped_column(unique=True)
1032 value: Mapped[str | None]
1033 created: Mapped[datetime] = mapped_column(
1034 DateTime(timezone=True), server_default=func.now()
1035 )
1036 modified: Mapped[datetime] = mapped_column(
1037 DateTime(timezone=True), server_default=func.now()
1038 )
1040 def __init__(self, *args, **kwargs):
1041 pass
1043 @override
1044 def __repr__(self):
1045 return "<DBConfig %s>" % self.name
1048__all__.append("DBConfig")
1050################################################################################
1053class DSCFile(Base):
1054 __tablename__ = "dsc_files"
1055 __table_args = (UniqueConstraint("source", "file", name="dsc_files_source_key"),)
1057 dscfile_id: Mapped[int] = mapped_column("id", primary_key=True)
1058 source_id: Mapped[int] = mapped_column("source", ForeignKey("source.id"))
1059 poolfile_id: Mapped[int] = mapped_column("file", ForeignKey("files.id"))
1060 created: Mapped[datetime | None] = mapped_column(
1061 DateTime(timezone=True), server_default=func.now()
1062 )
1063 modified: Mapped[datetime | None] = mapped_column(
1064 DateTime(timezone=True), server_default=func.now()
1065 )
1066 extra_file: Mapped[bool] = mapped_column(default=True)
1068 source: Mapped["DBSource"] = relationship(back_populates="srcfiles")
1069 poolfile: Mapped["PoolFile"] = relationship()
1071 def __init__(self, *args, **kwargs):
1072 pass
1074 @override
1075 def __repr__(self):
1076 return "<DSCFile %s>" % self.dscfile_id
1079__all__.append("DSCFile")
1082@session_wrapper
1083def get_dscfiles(
1084 dscfile_id: int | None = None,
1085 source_id: int | None = None,
1086 poolfile_id: int | None = None,
1087 session: "Session | None" = None,
1088) -> list[DSCFile]:
1089 """
1090 Returns a list of DSCFiles which may be empty
1092 :param dscfile_id: the dscfile_id of the DSCFiles to find
1093 :param source_id: the source id related to the DSCFiles to find
1094 :param poolfile_id: the poolfile id related to the DSCFiles to find
1095 :return: Possibly empty list of DSCFiles
1096 """
1098 assert session is not None
1100 q = select(DSCFile)
1101 if dscfile_id is not None:
1102 q = q.filter_by(dscfile_id=dscfile_id)
1103 if source_id is not None:
1104 q = q.filter_by(source_id=source_id)
1105 if poolfile_id is not None:
1106 q = q.filter_by(poolfile_id=poolfile_id)
1107 return list(session.scalars(q))
1110__all__.append("get_dscfiles")
1112################################################################################
1115class ExternalOverride(ORMObject):
1116 __tablename__ = "external_overrides"
1118 suite_id: Mapped[int] = mapped_column(
1119 "suite", ForeignKey("suite.id"), primary_key=True
1120 )
1121 component_id: Mapped[int] = mapped_column(
1122 "component", ForeignKey("component.id"), primary_key=True
1123 )
1124 package: Mapped[str] = mapped_column(primary_key=True)
1125 key: Mapped[str] = mapped_column(primary_key=True)
1126 value: Mapped[str]
1128 suite: Mapped["Suite"] = relationship()
1129 component: Mapped["Component"] = relationship()
1131 def __init__(self, *args, **kwargs):
1132 pass
1134 @override
1135 def __repr__(self):
1136 return "<ExternalOverride %s = %s: %s>" % (self.package, self.key, self.value)
1139__all__.append("ExternalOverride")
1141################################################################################
1144class PoolFile(ORMObject):
1145 __tablename__ = "files"
1147 file_id: Mapped[int] = mapped_column("id", primary_key=True)
1148 filename: Mapped[str] # TODO: add unique=True?
1149 filesize: Mapped[int] = mapped_column("size", BigInteger())
1150 md5sum: Mapped[str]
1151 last_used: Mapped[datetime | None]
1152 sha1sum: Mapped[str]
1153 sha256sum: Mapped[str]
1154 created: Mapped[datetime | None] = mapped_column(
1155 DateTime(timezone=True), server_default=func.now()
1156 )
1157 modified: Mapped[datetime | None] = mapped_column(
1158 DateTime(timezone=True), server_default=func.now()
1159 )
1161 archives: Mapped[list["ArchiveFile"]] = relationship(back_populates="file")
1163 def __init__(self, filename=None, filesize=-1, md5sum=None):
1164 self.filename = filename
1165 self.filesize = filesize
1166 self.md5sum = md5sum
1168 @property
1169 def fullpath(self) -> str:
1170 session = object_session(self)
1171 assert session is not None
1172 q = (
1173 select(ArchiveFile)
1174 .join(Archive)
1175 .where(ArchiveFile.file == self)
1176 .order_by(Archive.tainted.desc())
1177 .limit(1)
1178 )
1179 af = session.scalars(q).first()
1180 assert af is not None
1181 return af.path
1183 @property
1184 def component(self) -> Component:
1185 session = object_session(self)
1186 assert session is not None
1187 component_id = session.execute(
1188 select(ArchiveFile.component_id)
1189 .where(ArchiveFile.file == self)
1190 .group_by(ArchiveFile.component_id)
1191 ).scalar_one()
1192 return session.get_one(Component, component_id)
1194 @property
1195 def basename(self) -> str:
1196 return os.path.basename(self.filename)
1198 @override
1199 def properties(self) -> list[str]:
1200 return [
1201 "filename",
1202 "file_id",
1203 "filesize",
1204 "md5sum",
1205 "sha1sum",
1206 "sha256sum",
1207 "source",
1208 "binary",
1209 "last_used",
1210 ]
1213__all__.append("PoolFile")
1215################################################################################
1218class Fingerprint(ORMObject):
1219 __tablename__ = "fingerprint"
1221 fingerprint_id: Mapped[int] = mapped_column("id", primary_key=True)
1222 fingerprint: Mapped[str] = mapped_column(unique=True)
1223 uid_id: Mapped[int | None] = mapped_column("uid", ForeignKey("uid.id"))
1224 keyring_id: Mapped[int | None] = mapped_column("keyring", ForeignKey("keyrings.id"))
1225 created: Mapped[datetime | None] = mapped_column(
1226 DateTime(timezone=True), server_default=func.now()
1227 )
1228 modified: Mapped[datetime | None] = mapped_column(
1229 DateTime(timezone=True), server_default=func.now()
1230 )
1231 acl_id: Mapped[int | None] = mapped_column(ForeignKey("acl.id"))
1233 uid: Mapped["Uid | None"] = relationship(back_populates="fingerprint")
1234 keyring: Mapped["Keyring | None"] = relationship()
1235 acl: Mapped["ACL | None"] = relationship()
1237 def __init__(self, fingerprint=None):
1238 self.fingerprint = fingerprint
1240 @override
1241 def properties(self) -> list[str]:
1242 return ["fingerprint", "fingerprint_id", "keyring", "uid", "binary_reject"]
1245__all__.append("Fingerprint")
1248@session_wrapper
1249def get_fingerprint(fpr: str, session: "Session | None" = None) -> Fingerprint | None:
1250 """
1251 Returns Fingerprint object for given fpr.
1253 :param fpr: The fpr to find / add
1254 :param session: Optional SQL session object (a temporary one will be
1255 generated if not supplied).
1256 :return: the Fingerprint object for the given fpr or None
1257 """
1259 assert session is not None
1260 q = select(Fingerprint).where(Fingerprint.fingerprint == fpr)
1261 return session.execute(q).scalar_one_or_none()
1264__all__.append("get_fingerprint")
1267@session_wrapper
1268def get_or_set_fingerprint(fpr: str, session: "Session | None" = None) -> Fingerprint:
1269 """
1270 Returns Fingerprint object for given fpr.
1272 If no matching fpr is found, a row is inserted.
1274 :param fpr: The fpr to find / add
1275 :param session: Optional SQL session object (a temporary one will be
1276 generated if not supplied). If not passed, a commit will be performed at
1277 the end of the function, otherwise the caller is responsible for commiting.
1278 A flush will be performed either way.
1279 :return: the Fingerprint object for the given fpr
1280 """
1282 assert session is not None
1283 q = select(Fingerprint).where(Fingerprint.fingerprint == fpr)
1285 ret = session.execute(q).scalar_one_or_none()
1286 if ret is None:
1287 fingerprint = Fingerprint()
1288 fingerprint.fingerprint = fpr
1289 session.add(fingerprint)
1290 session.commit_or_flush() # type: ignore[attr-defined]
1291 ret = fingerprint
1293 return ret
1296__all__.append("get_or_set_fingerprint")
1298################################################################################
1300# Helper routine for Keyring class
1303def get_ldap_name(entry) -> str:
1304 name = []
1305 for k in ["cn", "mn", "sn"]:
1306 ret = entry.get(k)
1307 if not ret:
1308 continue
1309 value = ret[0].decode()
1310 if value and value[0] != "-":
1311 name.append(value)
1312 return " ".join(name)
1315################################################################################
1318class _Key(TypedDict, total=False):
1319 email: str
1320 name: str
1321 fingerprints: list[str]
1322 uid: str
1325class Keyring(Base):
1326 __tablename__ = "keyrings"
1328 keyring_id: Mapped[int] = mapped_column("id", primary_key=True)
1329 keyring_name: Mapped[str] = mapped_column("name", unique=True)
1330 priority: Mapped[int] = mapped_column(default=100)
1331 created: Mapped[datetime | None] = mapped_column(
1332 DateTime(timezone=True), server_default=func.now()
1333 )
1334 modified: Mapped[datetime | None] = mapped_column(
1335 DateTime(timezone=True), server_default=func.now()
1336 )
1337 active: Mapped[bool | None] = mapped_column(default=True)
1338 acl_id: Mapped[int | None] = mapped_column(ForeignKey("acl.id"))
1339 tag2upload: Mapped[bool] = mapped_column(default=False)
1341 acl: Mapped["ACL | None"] = relationship(foreign_keys=[acl_id])
1343 keys: dict[str, _Key] = {}
1344 fpr_lookup: dict[str, str] = {}
1346 def __init__(self, *args, **kwargs):
1347 pass
1349 @override
1350 def __repr__(self):
1351 return "<Keyring %s>" % self.keyring_name
1353 def de_escape_gpg_str(self, txt: str) -> str:
1354 esclist = re.split(r"(\\x..)", txt)
1355 for x in range(1, len(esclist), 2): 1355 ↛ 1356line 1355 didn't jump to line 1356 because the loop on line 1355 never started
1356 esclist[x] = "%c" % (int(esclist[x][2:], 16))
1357 return "".join(esclist)
1359 def parse_address(self, uid: str) -> tuple[str, str]:
1360 """parses uid and returns a tuple of real name and email address"""
1361 import email.utils
1363 (name, address) = email.utils.parseaddr(uid)
1364 name = re.sub(r"\s*[(].*[)]", "", name)
1365 name = self.de_escape_gpg_str(name)
1366 if name == "":
1367 name = uid
1368 return (name, address)
1370 def load_keys(self, keyring: str) -> None:
1371 if not self.keyring_id: 1371 ↛ 1372line 1371 didn't jump to line 1372 because the condition on line 1371 was never true
1372 raise Exception("Must be initialized with database information")
1374 cmd = [
1375 "gpg",
1376 "--no-default-keyring",
1377 "--keyring",
1378 keyring,
1379 "--with-colons",
1380 "--fingerprint",
1381 "--fingerprint",
1382 ]
1383 p = subprocess.Popen(cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
1385 key = None
1386 need_fingerprint = False
1388 assert p.stdout is not None
1389 for line_raw in p.stdout:
1390 try:
1391 line = line_raw.decode()
1392 except UnicodeDecodeError:
1393 # Some old UIDs might not use UTF-8 encoding. We assume they
1394 # use latin1.
1395 line = line_raw.decode("latin1")
1396 field = line.split(":")
1397 if field[0] == "pub":
1398 key = field[4]
1399 self.keys[key] = {}
1400 (name, addr) = self.parse_address(field[9])
1401 if "@" in addr: 1401 ↛ 1402line 1401 didn't jump to line 1402 because the condition on line 1401 was never true
1402 self.keys[key]["email"] = addr
1403 self.keys[key]["name"] = name
1404 need_fingerprint = True
1405 elif key and field[0] == "uid":
1406 assert key is not None
1407 (name, addr) = self.parse_address(field[9])
1408 if "email" not in self.keys[key] and "@" in addr: 1408 ↛ 1389line 1408 didn't jump to line 1389 because the condition on line 1408 was always true
1409 self.keys[key]["email"] = addr
1410 self.keys[key]["name"] = name
1411 elif need_fingerprint and field[0] == "fpr":
1412 assert key is not None
1413 self.keys[key]["fingerprints"] = [field[9]]
1414 self.fpr_lookup[field[9]] = key
1415 need_fingerprint = False
1417 (out, err) = p.communicate()
1418 r = p.returncode
1419 if r != 0: 1419 ↛ 1420line 1419 didn't jump to line 1420 because the condition on line 1419 was never true
1420 raise daklib.gpg.GpgException(
1421 "command failed: %s\nstdout: %r\nstderr: %r\n" % (cmd, out, err)
1422 )
1424 def import_users_from_ldap(
1425 self, session: "Session"
1426 ) -> tuple[dict[str, tuple[int, str]], dict[int, tuple[str, str]]]:
1427 import ldap # type: ignore
1429 from .utils import open_ldap_connection
1431 conn = open_ldap_connection()
1432 cnf = Config()
1433 LDAPDn = cnf["Import-LDAP-Fingerprints::LDAPDn"]
1434 Attrs = conn.search_s(
1435 LDAPDn,
1436 ldap.SCOPE_ONELEVEL,
1437 "(&(keyfingerprint=*)(supplementaryGid=%s))"
1438 % (cnf["Import-Users-From-Passwd::ValidGID"]),
1439 ["uid", "keyfingerprint", "cn", "mn", "sn"],
1440 )
1442 byuid: dict[int, tuple[str, str]] = {}
1443 byname: dict[str, tuple[int, str]] = {}
1445 for i in Attrs:
1446 entry = i[1]
1447 uid = entry["uid"][0].decode()
1448 name = get_ldap_name(entry)
1449 fingerprints = entry["keyFingerPrint"]
1450 keyid = None
1451 for f_raw in fingerprints:
1452 f = f_raw.decode()
1453 key = self.fpr_lookup.get(f, None)
1454 if key not in self.keys:
1455 continue
1456 self.keys[key]["uid"] = uid
1458 if keyid is not None:
1459 continue
1460 keyid = get_or_set_uid(uid, session).uid_id
1461 byuid[keyid] = (uid, name)
1462 byname[uid] = (keyid, name)
1464 return (byname, byuid)
1466 def generate_users_from_keyring(
1467 self, format: str, session: "Session"
1468 ) -> tuple[dict[str, tuple[int, str]], dict[int, tuple[str, str]]]:
1469 byuid: dict[int, tuple[str, str]] = {}
1470 byname: dict[str, tuple[int, str]] = {}
1471 any_invalid = False
1472 for x in list(self.keys.keys()):
1473 if "email" not in self.keys[x]: 1473 ↛ 1474line 1473 didn't jump to line 1474 because the condition on line 1473 was never true
1474 any_invalid = True
1475 self.keys[x]["uid"] = format % "invalid-uid"
1476 else:
1477 uid = format % self.keys[x]["email"]
1478 keyid = get_or_set_uid(uid, session).uid_id
1479 byuid[keyid] = (uid, self.keys[x]["name"])
1480 byname[uid] = (keyid, self.keys[x]["name"])
1481 self.keys[x]["uid"] = uid
1483 if any_invalid: 1483 ↛ 1484line 1483 didn't jump to line 1484 because the condition on line 1483 was never true
1484 uid = format % "invalid-uid"
1485 keyid = get_or_set_uid(uid, session).uid_id
1486 byuid[keyid] = (uid, "ungeneratable user id")
1487 byname[uid] = (keyid, "ungeneratable user id")
1489 return (byname, byuid)
1492__all__.append("Keyring")
1495@session_wrapper
1496def get_keyring(keyring: str, session: "Session | None" = None) -> Keyring | None:
1497 """
1498 If `keyring` does not have an entry in the `keyrings` table yet, return None
1499 If `keyring` already has an entry, simply return the existing :class:`Keyring`
1501 :param keyring: the keyring name
1502 :return: the :class:`Keyring` object for this keyring
1503 """
1505 assert session is not None
1506 q = select(Keyring).where(Keyring.keyring_name == keyring)
1507 return session.execute(q).scalar_one_or_none()
1510__all__.append("get_keyring")
1513@session_wrapper
1514def get_active_keyring_paths(session: "Session | None" = None) -> list[str]:
1515 """
1516 :return: list of active keyring paths
1517 """
1518 assert session is not None
1519 q = (
1520 select(Keyring.keyring_name)
1521 .where(Keyring.active.is_(True))
1522 .order_by(desc(Keyring.priority))
1523 )
1524 return list(session.scalars(q))
1527__all__.append("get_active_keyring_paths")
1530@session_wrapper
1531def get_active_keyring_files(session: "Session | None" = None) -> list[str]:
1532 """
1533 :return: list of active keyring file names, by ascending priority
1535 Note: get_active_keyring_paths() returns the same names by *descending*
1536 priority.
1537 """
1538 assert session is not None
1539 q = (
1540 select(Keyring.keyring_name)
1541 .where(Keyring.active.is_(True))
1542 .order_by(Keyring.priority)
1543 )
1544 return list(session.scalars(q))
1547__all__.append("get_active_keyring_files")
1549################################################################################
1552class DBChange(Base):
1553 __tablename__ = "changes"
1555 change_id: Mapped[int] = mapped_column("id", primary_key=True)
1556 changesname: Mapped[str]
1557 seen: Mapped[datetime] = mapped_column(
1558 DateTime(timezone=True), server_default=func.now()
1559 )
1560 source: Mapped[str]
1561 binaries: Mapped[str | None]
1562 architecture: Mapped[str]
1563 version: Mapped[str]
1564 distribution: Mapped[str]
1565 urgency: Mapped[str]
1566 maintainer: Mapped[str]
1567 fingerprint: Mapped[str]
1568 changedby: Mapped[str]
1569 date: Mapped[str]
1570 created: Mapped[datetime] = mapped_column(
1571 DateTime(timezone=True), server_default=func.now()
1572 )
1573 modified: Mapped[datetime] = mapped_column(
1574 DateTime(timezone=True), server_default=func.now()
1575 )
1576 changelog_id: Mapped[int | None]
1577 closes: Mapped[list[str] | None] = mapped_column(ARRAY(String()))
1578 authorized_by_fingerprint: Mapped[str | None]
1580 def __init__(self, *args, **kwargs):
1581 pass
1583 @override
1584 def __repr__(self):
1585 return "<DBChange %s>" % self.changesname
1588__all__.append("DBChange")
1591@session_wrapper
1592def get_dbchange(filename: str, session: "Session | None" = None) -> DBChange | None:
1593 """
1594 returns DBChange object for given `filename`.
1596 :param filename: the name of the file
1597 :param session: Optional SQLA session object (a temporary one will be
1598 generated if not supplied)
1599 :return: DBChange object for the given filename (:const:`None` if not present)
1600 """
1601 assert session is not None
1602 q = select(DBChange).where(DBChange.changesname == filename)
1603 return session.execute(q).scalar_one_or_none()
1606__all__.append("get_dbchange")
1608################################################################################
1611class DBChangelog(Base):
1612 __tablename__ = "changelogs_text"
1614 id: Mapped[int] = mapped_column(primary_key=True)
1615 changelog: Mapped[str | None]
1617 def __init__(self, *args, **kwargs):
1618 pass
1620 @override
1621 def __repr__(self):
1622 return "<DBChangelog %s>" % self.id
1625__all__.append("DBChangelog")
1627################################################################################
1630class Maintainer(ORMObject):
1631 __tablename__ = "maintainer"
1633 maintainer_id: Mapped[int] = mapped_column("id", primary_key=True)
1634 name: Mapped[str] = mapped_column(unique=True)
1635 created: Mapped[datetime | None] = mapped_column(
1636 DateTime(timezone=True), server_default=func.now()
1637 )
1638 modified: Mapped[datetime | None] = mapped_column(
1639 DateTime(timezone=True), server_default=func.now()
1640 )
1642 maintains_sources: Mapped[list["DBSource"]] = relationship(
1643 back_populates="maintainer", foreign_keys=lambda: DBSource.maintainer_id
1644 )
1645 changed_sources: Mapped[list["DBSource"]] = relationship(
1646 back_populates="changedby", foreign_keys=lambda: DBSource.changedby_id
1647 )
1649 def __init__(self, name=None):
1650 self.name = name
1652 @override
1653 def properties(self) -> list[str]:
1654 return ["name", "maintainer_id"]
1656 def get_split_maintainer(self) -> tuple[str, str, str, str]:
1657 if not hasattr(self, "name") or self.name is None:
1658 return ("", "", "", "")
1660 return fix_maintainer(self.name.strip())
1663__all__.append("Maintainer")
1666@session_wrapper
1667def get_or_set_maintainer(name: str, session: "Session | None" = None) -> Maintainer:
1668 """
1669 Returns Maintainer object for given maintainer name.
1671 If no matching maintainer name is found, a row is inserted.
1673 :param name: The maintainer name to add
1674 :param session: Optional SQL session object (a temporary one will be
1675 generated if not supplied). If not passed, a commit will be performed at
1676 the end of the function, otherwise the caller is responsible for commiting.
1677 A flush will be performed either way.
1678 :return: the Maintainer object for the given maintainer
1679 """
1681 assert session is not None
1682 q = select(Maintainer).where(Maintainer.name == name)
1683 ret = session.execute(q).scalar_one_or_none()
1684 if ret is None:
1685 maintainer = Maintainer()
1686 maintainer.name = name
1687 session.add(maintainer)
1688 session.commit_or_flush() # type: ignore[attr-defined]
1689 ret = maintainer
1691 return ret
1694__all__.append("get_or_set_maintainer")
1697@session_wrapper
1698def get_maintainer(
1699 maintainer_id: int, session: "Session | None" = None
1700) -> Maintainer | None:
1701 """
1702 Return the name of the maintainer behind `maintainer_id` or :const:`None`
1703 if that `maintainer_id` is invalid.
1705 :param maintainer_id: the id of the maintainer
1706 :return: the Maintainer with this `maintainer_id`
1707 """
1709 assert session is not None
1710 return session.get(Maintainer, maintainer_id)
1713__all__.append("get_maintainer")
1715################################################################################
1718class NewComment(Base):
1719 __tablename__ = "new_comments"
1721 comment_id: Mapped[int] = mapped_column("id", primary_key=True)
1722 package: Mapped[str]
1723 version: Mapped[str]
1724 comment: Mapped[str]
1725 author: Mapped[str]
1726 notedate: Mapped[datetime] = mapped_column(
1727 DateTime(timezone=True), server_default=func.now()
1728 )
1729 trainee: Mapped[bool] = mapped_column(default=False)
1730 created: Mapped[datetime | None] = mapped_column(
1731 DateTime(timezone=True), server_default=func.now()
1732 )
1733 modified: Mapped[datetime | None] = mapped_column(
1734 DateTime(timezone=True), server_default=func.now()
1735 )
1736 policy_queue_id: Mapped[int] = mapped_column(ForeignKey("policy_queue.id"))
1738 policy_queue: Mapped["PolicyQueue"] = relationship()
1740 def __init__(self, *args, **kwargs):
1741 pass
1743 @override
1744 def __repr__(self):
1745 return """<NewComment for '%s %s' (%s)>""" % (
1746 self.package,
1747 self.version,
1748 self.comment_id,
1749 )
1752__all__.append("NewComment")
1755@session_wrapper
1756def has_new_comment(
1757 policy_queue: "PolicyQueue",
1758 package: str,
1759 version: str,
1760 session: "Session | None" = None,
1761) -> bool:
1762 """
1763 Returns :const:`True` if the given combination of `package`, `version` has a comment.
1765 :param package: name of the package
1766 :param version: package version
1767 :param session: Optional SQLA session object (a temporary one will be
1768 generated if not supplied)
1769 """
1771 assert session is not None
1772 q = select(
1773 exists().where(
1774 NewComment.policy_queue == policy_queue,
1775 NewComment.package == package,
1776 NewComment.version == version,
1777 )
1778 )
1779 return bool(session.scalar(q))
1782__all__.append("has_new_comment")
1785@session_wrapper
1786def get_new_comments(
1787 policy_queue: "PolicyQueue",
1788 package: str | None = None,
1789 version: str | None = None,
1790 comment_id: int | None = None,
1791 session: "Session | None" = None,
1792) -> list[NewComment]:
1793 """
1794 Returns (possibly empty) list of NewComment objects for the given
1795 parameters
1797 :param package: name of the package
1798 :param version: package version
1799 :param comment_id: An id of a comment
1800 :param session: Optional SQLA session object (a temporary one will be
1801 generated if not supplied)
1802 :return: A (possibly empty) list of NewComment objects will be returned
1803 """
1805 assert session is not None
1806 q = select(NewComment).filter_by(policy_queue=policy_queue)
1807 if package is not None: 1807 ↛ 1809line 1807 didn't jump to line 1809 because the condition on line 1807 was always true
1808 q = q.filter_by(package=package)
1809 if version is not None: 1809 ↛ 1810line 1809 didn't jump to line 1810 because the condition on line 1809 was never true
1810 q = q.filter_by(version=version)
1811 if comment_id is not None: 1811 ↛ 1812line 1811 didn't jump to line 1812 because the condition on line 1811 was never true
1812 q = q.filter_by(comment_id=comment_id)
1814 return list(session.scalars(q))
1817__all__.append("get_new_comments")
1819################################################################################
1822class Override(ORMObject):
1823 __tablename__ = "override"
1825 suite_id: Mapped[int] = mapped_column(
1826 "suite", ForeignKey("suite.id"), primary_key=True
1827 )
1828 component_id: Mapped[int] = mapped_column(
1829 "component", ForeignKey("component.id"), primary_key=True
1830 )
1831 package: Mapped[str] = mapped_column(primary_key=True)
1832 overridetype_id: Mapped[int] = mapped_column(
1833 "type", ForeignKey("override_type.id"), primary_key=True
1834 )
1835 priority_id: Mapped[int | None] = mapped_column(
1836 "priority", ForeignKey("priority.id")
1837 )
1838 section_id: Mapped[int] = mapped_column("section", ForeignKey("section.id"))
1839 maintainer: Mapped[str | None]
1840 created: Mapped[datetime | None] = mapped_column(
1841 DateTime(timezone=True), server_default=func.now()
1842 )
1843 modified: Mapped[datetime | None] = mapped_column(
1844 DateTime(timezone=True), server_default=func.now()
1845 )
1847 suite: Mapped["Suite"] = relationship(back_populates="overrides")
1848 component: Mapped["Component"] = relationship(back_populates="overrides")
1849 priority: Mapped["Priority"] = relationship(back_populates="overrides")
1850 section: Mapped["Section"] = relationship(back_populates="overrides")
1851 overridetype: Mapped["OverrideType"] = relationship(back_populates="overrides")
1853 def __init__(
1854 self,
1855 package=None,
1856 suite=None,
1857 component=None,
1858 overridetype=None,
1859 section=None,
1860 priority=None,
1861 ):
1862 self.package = package
1863 self.suite = suite
1864 self.component = component
1865 self.overridetype = overridetype
1866 self.section = section
1867 self.priority = priority
1869 @override
1870 def properties(self) -> list[str]:
1871 return ["package", "suite", "component", "overridetype", "section", "priority"]
1874__all__.append("Override")
1877@session_wrapper
1878def get_override(
1879 package: str,
1880 suite: str | list[str] | None = None,
1881 component: str | list[str] | None = None,
1882 overridetype: str | list[str] | None = None,
1883 session: "Session | None" = None,
1884) -> list[Override]:
1885 """
1886 Returns Override object for the given parameters
1888 :param package: The name of the package
1889 :param suite: The name of the suite (or suites if a list) to limit to. If
1890 None, don't limit. Defaults to None.
1891 :param component: The name of the component (or components if a list) to
1892 limit to. If None, don't limit. Defaults to None.
1893 :param overridetype: The name of the overridetype (or overridetypes if a list) to
1894 limit to. If None, don't limit. Defaults to None.
1895 :param session: Optional SQLA session object (a temporary one will be
1896 generated if not supplied)
1897 :return: A (possibly empty) list of Override objects will be returned
1898 """
1900 assert session is not None
1902 q = select(Override).filter_by(package=package)
1904 if suite is not None:
1905 if not isinstance(suite, list):
1906 suite = [suite]
1907 q = q.join(Suite).where(Suite.suite_name.in_(suite))
1909 if component is not None:
1910 if not isinstance(component, list):
1911 component = [component]
1912 q = q.join(Component).where(Component.component_name.in_(component))
1914 if overridetype is not None:
1915 if not isinstance(overridetype, list):
1916 overridetype = [overridetype]
1917 q = q.join(OverrideType).where(OverrideType.overridetype.in_(overridetype))
1919 return list(session.scalars(q))
1922__all__.append("get_override")
1925################################################################################
1928class OverrideType(ORMObject):
1929 __tablename__ = "override_type"
1931 overridetype_id: Mapped[int] = mapped_column("id", primary_key=True)
1932 overridetype: Mapped[str] = mapped_column("type", unique=True)
1933 created: Mapped[datetime | None] = mapped_column(
1934 DateTime(timezone=True), server_default=func.now()
1935 )
1936 modified: Mapped[datetime | None] = mapped_column(
1937 DateTime(timezone=True), server_default=func.now()
1938 )
1940 overrides: DynamicMapped["Override"] = relationship(back_populates="overridetype")
1942 def __init__(self, overridetype=None):
1943 self.overridetype = overridetype
1945 @override
1946 def properties(self) -> list[str]:
1947 return ["overridetype", "overridetype_id", "overrides_count"]
1950__all__.append("OverrideType")
1953@session_wrapper
1954def get_override_type(
1955 override_type: str, session: "Session | None" = None
1956) -> OverrideType | None:
1957 """
1958 Returns OverrideType object for given `override_type`.
1960 :param override_type: The name of the override type
1961 :param session: Optional SQLA session object (a temporary one will be
1962 generated if not supplied)
1963 :return: the database id for the given override type
1964 """
1966 assert session is not None
1967 q = select(OverrideType).where(OverrideType.overridetype == override_type)
1968 return session.execute(q).scalar_one_or_none()
1971__all__.append("get_override_type")
1973################################################################################
1976class PolicyQueue(Base):
1977 __tablename__ = "policy_queue"
1979 policy_queue_id: Mapped[int] = mapped_column("id", primary_key=True)
1980 queue_name: Mapped[str] = mapped_column(unique=True)
1981 path: Mapped[str]
1982 change_perms: Mapped[str] = mapped_column(CHAR(length=4))
1983 generate_metadata: Mapped[bool] = mapped_column(default=False)
1984 created: Mapped[datetime | None] = mapped_column(
1985 DateTime(timezone=True), server_default=func.now()
1986 )
1987 modified: Mapped[datetime | None] = mapped_column(
1988 DateTime(timezone=True), server_default=func.now()
1989 )
1990 send_to_build_queues: Mapped[bool] = mapped_column(default=False)
1991 suite_id: Mapped[int] = mapped_column(ForeignKey("suite.id"))
1993 suite: Mapped["Suite"] = relationship(foreign_keys=[suite_id])
1994 uploads: Mapped[list["PolicyQueueUpload"]] = relationship(
1995 back_populates="policy_queue"
1996 )
1998 def __init__(self, *args, **kwargs):
1999 pass
2001 @override
2002 def __repr__(self):
2003 return "<PolicyQueue %s>" % self.queue_name
2006__all__.append("PolicyQueue")
2009@session_wrapper
2010def get_policy_queue(
2011 queuename: str, session: "Session | None" = None
2012) -> PolicyQueue | None:
2013 """
2014 Returns PolicyQueue object for given `queuename`
2016 :param queuename: The name of the queue
2017 :param session: Optional SQLA session object (a temporary one will be
2018 generated if not supplied)
2019 :return: PolicyQueue object for the given queue
2020 """
2022 assert session is not None
2023 q = select(PolicyQueue).where(PolicyQueue.queue_name == queuename)
2024 return session.execute(q).scalar_one_or_none()
2027__all__.append("get_policy_queue")
2029################################################################################
2032@functools.total_ordering
2033class PolicyQueueUpload(Base):
2034 __tablename__ = "policy_queue_upload"
2035 __table_args = (
2036 UniqueConstraint(
2037 "policy_queue_id",
2038 "target_suite_id",
2039 "changes_id",
2040 name="policy_queue_upload_policy_queue_id_target_suite_id_changes_key",
2041 ),
2042 )
2044 id: Mapped[int] = mapped_column(primary_key=True)
2045 policy_queue_id: Mapped[int] = mapped_column(ForeignKey("policy_queue.id"))
2046 target_suite_id: Mapped[int] = mapped_column(ForeignKey("suite.id"))
2047 changes_id: Mapped[int] = mapped_column(ForeignKey("changes.id"))
2048 source_id: Mapped[int | None] = mapped_column(ForeignKey("source.id"))
2050 changes: Mapped["DBChange"] = relationship()
2051 policy_queue: Mapped["PolicyQueue"] = relationship(back_populates="uploads")
2052 target_suite: Mapped["Suite"] = relationship()
2053 source: Mapped["DBSource | None"] = relationship()
2054 binaries: Mapped[list["DBBinary"]] = relationship(
2055 secondary="policy_queue_upload_binaries_map"
2056 )
2057 byhand: Mapped[list["PolicyQueueByhandFile"]] = relationship(
2058 back_populates="upload"
2059 )
2061 def _key(self):
2062 return (
2063 self.changes.source,
2064 AptVersion(self.changes.version),
2065 self.source is None,
2066 self.changes.changesname,
2067 )
2069 @override
2070 def __eq__(self, other: object) -> bool:
2071 if not isinstance(other, PolicyQueueUpload): 2071 ↛ 2072line 2071 didn't jump to line 2072 because the condition on line 2071 was never true
2072 return NotImplemented
2073 return self._key() == other._key()
2075 def __lt__(self, other):
2076 return self._key() < other._key()
2079__all__.append("PolicyQueueUpload")
2082class PolicyQueueUploadBinariesMap(Base):
2083 __tablename__ = "policy_queue_upload_binaries_map"
2085 policy_queue_upload_id: Mapped[int] = mapped_column(
2086 ForeignKey("policy_queue_upload.id", ondelete="CASCADE"), primary_key=True
2087 )
2088 binary_id: Mapped[int] = mapped_column(ForeignKey("binaries.id"), primary_key=True)
2091################################################################################
2094class PolicyQueueByhandFile(Base):
2095 __tablename__ = "policy_queue_byhand_file"
2097 id: Mapped[int] = mapped_column(primary_key=True)
2098 upload_id: Mapped[int] = mapped_column(ForeignKey("policy_queue_upload.id"))
2099 filename: Mapped[str]
2100 processed: Mapped[bool] = mapped_column(default=False)
2102 upload: Mapped[PolicyQueueUpload] = relationship(back_populates="byhand")
2105__all__.append("PolicyQueueByhandFile")
2107################################################################################
2110class Priority(ORMObject):
2111 __tablename__ = "priority"
2113 priority_id: Mapped[int] = mapped_column("id", primary_key=True)
2114 priority: Mapped[str] = mapped_column(unique=True)
2115 level: Mapped[int] = mapped_column(unique=True)
2116 created: Mapped[datetime | None] = mapped_column(
2117 DateTime(timezone=True), server_default=func.now()
2118 )
2119 modified: Mapped[datetime | None] = mapped_column(
2120 DateTime(timezone=True), server_default=func.now()
2121 )
2123 overrides: DynamicMapped["Override"] = relationship(back_populates="priority")
2125 def __init__(self, priority=None, level=None):
2126 self.priority = priority
2127 self.level = level
2129 @override
2130 def properties(self) -> list[str]:
2131 return ["priority", "priority_id", "level", "overrides_count"]
2133 @override
2134 def __eq__(self, val: object):
2135 if isinstance(val, str):
2136 warnings.warn(
2137 "comparison with a `str` is deprecated",
2138 DeprecationWarning,
2139 stacklevel=2,
2140 )
2141 return self.priority == val
2142 # This signals to use the normal comparison operator
2143 return NotImplemented
2145 @override
2146 def __ne__(self, val: object):
2147 if isinstance(val, str): 2147 ↛ 2155line 2147 didn't jump to line 2155 because the condition on line 2147 was always true
2148 warnings.warn(
2149 "comparison with a `str` is deprecated",
2150 DeprecationWarning,
2151 stacklevel=2,
2152 )
2153 return self.priority != val
2154 # This signals to use the normal comparison operator
2155 return NotImplemented
2157 __hash__ = ORMObject.__hash__
2160__all__.append("Priority")
2163@session_wrapper
2164def get_priority(priority: str, session: "Session | None" = None) -> Priority | None:
2165 """
2166 Returns Priority object for given `priority` name.
2168 :param priority: The name of the priority
2169 :param session: Optional SQLA session object (a temporary one will be
2170 generated if not supplied)
2171 :return: Priority object for the given priority
2172 """
2174 assert session is not None
2175 q = select(Priority).where(Priority.priority == priority)
2176 return session.execute(q).scalar_one_or_none()
2179__all__.append("get_priority")
2182@session_wrapper
2183def get_priorities(session: "Session | None" = None) -> dict[str, int]:
2184 """
2185 Returns dictionary of priority names -> id mappings
2187 :param session: Optional SQL session object (a temporary one will be
2188 generated if not supplied)
2189 :return: dictionary of priority names -> id mappings
2190 """
2192 assert session is not None
2193 q = select(Priority.priority, Priority.priority_id)
2194 return {priority: priority_id for priority, priority_id in session.execute(q)}
2197__all__.append("get_priorities")
2199################################################################################
2202class Section(BaseTimestamp):
2203 __tablename__ = "section"
2205 section_id: Mapped[int] = mapped_column("id", primary_key=True)
2206 section: Mapped[str] = mapped_column(unique=True)
2208 overrides: DynamicMapped["Override"] = relationship(back_populates="section")
2210 def __init__(self, section=None):
2211 self.section = section
2213 @override
2214 def __str__(self):
2215 return self.section
2217 @override
2218 def __repr__(self):
2219 return "<{} {}>".format(
2220 self.__class__.__name__,
2221 self.section,
2222 )
2224 @override
2225 def __eq__(self, val: object):
2226 if isinstance(val, str):
2227 warnings.warn(
2228 "comparison with a `str` is deprecated",
2229 DeprecationWarning,
2230 stacklevel=2,
2231 )
2232 return self.section == val
2233 # This signals to use the normal comparison operator
2234 return NotImplemented
2236 @override
2237 def __ne__(self, val: object):
2238 if isinstance(val, str): 2238 ↛ 2246line 2238 didn't jump to line 2246 because the condition on line 2238 was always true
2239 warnings.warn(
2240 "comparison with a `str` is deprecated",
2241 DeprecationWarning,
2242 stacklevel=2,
2243 )
2244 return self.section != val
2245 # This signals to use the normal comparison operator
2246 return NotImplemented
2248 __hash__ = BaseTimestamp.__hash__
2251__all__.append("Section")
2254@session_wrapper
2255def get_section(section: str, session: "Session | None" = None) -> Section | None:
2256 """
2257 Returns Section object for given `section` name.
2259 :param section: The name of the section
2260 :param session: Optional SQLA session object (a temporary one will be
2261 generated if not supplied)
2262 :return: Section object for the given section name
2263 """
2265 assert session is not None
2266 q = select(Section).where(Section.section == section)
2267 return session.execute(q).scalar_one_or_none()
2270__all__.append("get_section")
2273@session_wrapper
2274def get_sections(session: "Session | None" = None) -> dict[str, int]:
2275 """
2276 Returns dictionary of section names -> id mappings
2278 :param session: Optional SQL session object (a temporary one will be
2279 generated if not supplied)
2280 :return: dictionary of section names -> id mappings
2281 """
2283 assert session is not None
2284 q = select(Section.section, Section.section_id)
2285 return {section: section_id for section, section_id in session.execute(q)}
2288__all__.append("get_sections")
2290################################################################################
2293class SignatureHistory(ORMObject):
2294 __tablename__ = "signature_history"
2296 fingerprint: Mapped[str] = mapped_column(primary_key=True)
2297 signature_timestamp: Mapped[datetime] = mapped_column(
2298 DateTime(timezone=False), primary_key=True
2299 )
2300 contents_sha1: Mapped[str] = mapped_column(primary_key=True)
2301 seen: Mapped[datetime] = mapped_column(
2302 DateTime(timezone=False), server_default=func.now()
2303 )
2305 @classmethod
2306 def from_signed_file(
2307 cls, signed_file: "daklib.gpg.SignedFile"
2308 ) -> "SignatureHistory":
2309 """signature history entry from signed file
2311 :param signed_file: signed file
2312 """
2313 self = cls()
2314 self.fingerprint = signed_file.primary_fingerprint
2315 self.signature_timestamp = signed_file.signature_timestamp
2316 self.contents_sha1 = signed_file.contents_sha1
2317 return self
2319 def query(self, session):
2320 q = (
2321 select(SignatureHistory)
2322 .filter_by(
2323 fingerprint=self.fingerprint,
2324 signature_timestamp=self.signature_timestamp,
2325 contents_sha1=self.contents_sha1,
2326 )
2327 .limit(1)
2328 )
2329 return session.scalars(q).first()
2332__all__.append("SignatureHistory")
2334################################################################################
2337class SrcContents(ORMObject):
2338 __tablename__ = "src_contents"
2340 source_id: Mapped[int] = mapped_column(
2341 ForeignKey("source.id", ondelete="CASCADE"), primary_key=True
2342 )
2343 file: Mapped[str] = mapped_column(primary_key=True)
2345 source: Mapped["DBSource"] = relationship(back_populates="contents")
2347 def __init__(self, file=None, source=None):
2348 self.file = file
2349 self.source = source
2351 @override
2352 def properties(self) -> list[str]:
2353 return ["file", "source"]
2356__all__.append("SrcContents")
2358################################################################################
2361class DBSource(ORMObject):
2362 __tablename__ = "source"
2363 __table_args = (UniqueConstraint("source", "version", name="source_source_key"),)
2365 source_id: Mapped[int] = mapped_column("id", primary_key=True)
2366 source: Mapped[str]
2367 version: Mapped[str] = mapped_column(DebVersion())
2368 maintainer_id: Mapped[int] = mapped_column(
2369 "maintainer", ForeignKey("maintainer.id")
2370 )
2371 poolfile_id: Mapped[int] = mapped_column(
2372 "file", ForeignKey("files.id"), unique=True
2373 )
2374 fingerprint_id: Mapped[int | None] = mapped_column(
2375 "sig_fpr", ForeignKey("fingerprint.id")
2376 )
2377 install_date: Mapped[datetime] = mapped_column(
2378 DateTime(timezone=True), server_default=func.now()
2379 )
2380 changedby_id: Mapped[int] = mapped_column("changedby", ForeignKey("maintainer.id"))
2381 dm_upload_allowed: Mapped[bool] = mapped_column(default=False)
2382 created: Mapped[datetime | None] = mapped_column(
2383 DateTime(timezone=True), server_default=func.now()
2384 )
2385 modified: Mapped[datetime | None] = mapped_column(
2386 DateTime(timezone=True), server_default=func.now()
2387 )
2388 stanza: Mapped[str]
2389 authorized_by_fingerprint_id: Mapped[int | None] = mapped_column(
2390 ForeignKey("fingerprint.id")
2391 )
2393 poolfile: Mapped["PoolFile"] = relationship()
2394 fingerprint: Mapped["Fingerprint | None"] = relationship(
2395 foreign_keys=[fingerprint_id]
2396 )
2397 authorized_by_fingerprint: Mapped["Fingerprint | None"] = relationship(
2398 foreign_keys=[authorized_by_fingerprint_id]
2399 )
2400 srcfiles: Mapped[list["DSCFile"]] = relationship(back_populates="source")
2401 suites: Mapped[list["Suite"]] = relationship(
2402 secondary="src_associations", back_populates="sources"
2403 )
2404 uploaders: Mapped[list["Maintainer"]] = relationship(secondary="src_uploaders")
2405 key: Mapped[dict["MetadataKey", "SourceMetadata"]] = relationship(
2406 collection_class=attribute_keyed_dict("key"),
2407 cascade="all",
2408 back_populates="source",
2409 )
2410 contents: DynamicMapped["SrcContents"] = relationship(
2411 cascade="all",
2412 back_populates="source",
2413 )
2414 binaries: Mapped[list["DBBinary"]] = relationship(back_populates="source")
2415 extra_binary_references: DynamicMapped["DBBinary"] = relationship(
2416 secondary="extra_src_references", back_populates="extra_sources"
2417 )
2418 maintainer: Mapped["Maintainer"] = relationship(
2419 back_populates="maintains_sources", foreign_keys=[maintainer_id]
2420 )
2421 changedby: Mapped["Maintainer"] = relationship(
2422 back_populates="changed_sources", foreign_keys=[changedby_id]
2423 )
2425 metadata_proxy: AssociationProxy[dict["MetadataKey", str]] = association_proxy(
2426 "key", "value"
2427 )
2429 def __init__(
2430 self,
2431 source=None,
2432 version=None,
2433 maintainer=None,
2434 changedby=None,
2435 poolfile=None,
2436 install_date=None,
2437 fingerprint=None,
2438 ):
2439 self.source = source
2440 self.version = version
2441 self.maintainer = maintainer
2442 self.changedby = changedby
2443 self.poolfile = poolfile
2444 self.install_date = install_date
2445 self.fingerprint = fingerprint
2447 @property
2448 def pkid(self) -> int:
2449 return self.source_id
2451 @property
2452 def name(self) -> str:
2453 return self.source
2455 @property
2456 def arch_string(self) -> str:
2457 return "source"
2459 @override
2460 def properties(self) -> list[str]:
2461 return [
2462 "source",
2463 "source_id",
2464 "maintainer",
2465 "changedby",
2466 "fingerprint",
2467 "poolfile",
2468 "version",
2469 "suites_count",
2470 "install_date",
2471 "binaries_count",
2472 "uploaders_count",
2473 ]
2475 def read_control_fields(self) -> Deb822:
2476 """
2477 Reads the control information from a dsc
2479 :return: fields is the dsc information in a dictionary form
2480 """
2481 with open(self.poolfile.fullpath, "r") as fd:
2482 fields = Deb822(fd)
2483 return fields
2485 def scan_contents(self) -> set[str]:
2486 """
2487 Returns a set of names for non directories. The path names are
2488 normalized after converting them from either utf-8 or iso8859-1
2489 encoding.
2490 """
2491 fullpath = self.poolfile.fullpath
2492 from daklib.contents import UnpackedSource
2494 unpacked = UnpackedSource(fullpath)
2495 fileset = set()
2496 for name in unpacked.get_all_filenames():
2497 fileset.add(name)
2498 return fileset
2500 @property
2501 def proxy(self) -> "MetadataProxy":
2502 session = object_session(self)
2503 assert session is not None
2504 query = select(SourceMetadata).filter_by(source=self)
2505 return MetadataProxy(session, query)
2508__all__.append("DBSource")
2511@session_wrapper
2512def get_suites_source_in(source: str, session=None) -> "list[Suite]":
2513 """
2514 Returns list of Suite objects which given `source` name is in
2516 :param source: DBSource package name to search for
2517 :return: list of Suite objects for the given source
2518 """
2520 q = select(Suite).where(Suite.sources.any(source=source))
2521 return list(session.scalars(q))
2524__all__.append("get_suites_source_in")
2526# FIXME: This function fails badly if it finds more than 1 source package and
2527# its implementation is trivial enough to be inlined.
2530@session_wrapper
2531def get_source_in_suite(
2532 source: str, suite_name: str | None, session: "Session | None" = None
2533) -> DBSource | None:
2534 """
2535 Returns a DBSource object for a combination of `source` and `suite_name`.
2537 :param source: source package name
2538 :param suite_name: the suite name
2539 :return: the version for `source` in `suite`
2540 """
2541 assert session is not None
2542 if suite_name is None: 2542 ↛ 2543line 2542 didn't jump to line 2543 because the condition on line 2542 was never true
2543 return None
2544 suite = get_suite(suite_name, session)
2545 if suite is None: 2545 ↛ 2546line 2545 didn't jump to line 2546 because the condition on line 2545 was never true
2546 return None
2547 return session.execute(suite.get_sources(source)).scalar_one_or_none()
2550__all__.append("get_source_in_suite")
2553@session_wrapper
2554def import_metadata_into_db(
2555 obj: DBBinary | DBSource, session: "Session | None" = None
2556) -> None:
2557 """
2558 This routine works on either DBBinary or DBSource objects and imports
2559 their metadata into the database
2560 """
2561 assert session is not None
2562 fields = obj.read_control_fields()
2563 for k in fields.keys():
2564 try:
2565 # Try raw ASCII
2566 val = str(fields[k])
2567 except UnicodeEncodeError:
2568 # Fall back to UTF-8
2569 try:
2570 val = fields[k].encode("utf-8")
2571 except UnicodeEncodeError:
2572 # Finally try iso8859-1
2573 val = fields[k].encode("iso8859-1")
2574 # Otherwise we allow the exception to percolate up and we cause
2575 # a reject as someone is playing silly buggers
2577 obj.metadata_proxy[get_or_set_metadatakey(k, session)] = val
2579 session.commit_or_flush() # type: ignore[attr-defined]
2582__all__.append("import_metadata_into_db")
2585class SrcAssociations(Base):
2586 __tablename__ = "src_associations"
2587 __table_args = (
2588 UniqueConstraint("suite", "source", name="src_associations_suite_key"),
2589 )
2591 id: Mapped[int] = mapped_column(primary_key=True)
2592 suite: Mapped[int] = mapped_column(ForeignKey("suite.id"))
2593 source: Mapped[int] = mapped_column(ForeignKey("source.id"))
2594 created: Mapped[datetime] = mapped_column(
2595 DateTime(timezone=True), server_default=func.now()
2596 )
2597 modified: Mapped[datetime] = mapped_column(
2598 DateTime(timezone=True), server_default=func.now()
2599 )
2602class SrcUploaders(Base):
2603 __tablename__ = "src_uploaders"
2604 __table_args = (
2605 UniqueConstraint("source", "maintainer", name="src_uploaders_source_key"),
2606 )
2608 id: Mapped[int] = mapped_column(primary_key=True)
2609 source: Mapped[int] = mapped_column(ForeignKey("source.id", ondelete="CASCADE"))
2610 maintainer: Mapped[int] = mapped_column(
2611 ForeignKey("maintainer.id", ondelete="CASCADE")
2612 )
2613 created: Mapped[datetime] = mapped_column(
2614 DateTime(timezone=True), server_default=func.now()
2615 )
2616 modified: Mapped[datetime] = mapped_column(
2617 DateTime(timezone=True), server_default=func.now()
2618 )
2621################################################################################
2624class SrcFormat(Base):
2625 __tablename__ = "src_format"
2627 src_format_id: Mapped[int] = mapped_column("id", primary_key=True)
2628 format_name: Mapped[str] = mapped_column(unique=True)
2629 created: Mapped[datetime] = mapped_column(
2630 DateTime(timezone=True), server_default=func.now()
2631 )
2632 modified: Mapped[datetime] = mapped_column(
2633 DateTime(timezone=True), server_default=func.now()
2634 )
2636 suites: DynamicMapped["Suite"] = relationship(
2637 secondary="suite_src_formats", back_populates="srcformats"
2638 )
2640 def __init__(self, *args, **kwargs):
2641 pass
2643 @override
2644 def __repr__(self):
2645 return "<SrcFormat %s>" % (self.format_name)
2648__all__.append("SrcFormat")
2650################################################################################
2652SUITE_FIELDS = [
2653 ("SuiteName", "suite_name"),
2654 ("SuiteID", "suite_id"),
2655 ("Version", "version"),
2656 ("Origin", "origin"),
2657 ("Label", "label"),
2658 ("Description", "description"),
2659 ("Untouchable", "untouchable"),
2660 ("Announce", "announce"),
2661 ("Codename", "codename"),
2662 ("OverrideCodename", "overridecodename"),
2663 ("ValidTime", "validtime"),
2664 ("Priority", "priority"),
2665 ("NotAutomatic", "notautomatic"),
2666 ("CopyChanges", "copychanges"),
2667 ("OverrideSuite", "overridesuite"),
2668]
2670# Why the heck don't we have any UNIQUE constraints in table suite?
2671# TODO: Add UNIQUE constraints for appropriate columns.
2674class Suite(ORMObject):
2675 __tablename__ = "suite"
2677 suite_id: Mapped[int] = mapped_column("id", primary_key=True)
2678 suite_name: Mapped[str] = mapped_column(unique=True)
2679 version: Mapped[str | None]
2680 origin: Mapped[str | None]
2681 label: Mapped[str | None]
2682 description: Mapped[str | None]
2683 untouchable: Mapped[bool] = mapped_column(default=False)
2684 codename: Mapped[str | None]
2685 overridecodename: Mapped[str | None]
2686 validtime: Mapped[int] = mapped_column(default=604800)
2687 priority: Mapped[int] = mapped_column(default=0)
2688 notautomatic: Mapped[bool] = mapped_column(default=False)
2689 copychanges: Mapped[str | None]
2690 overridesuite: Mapped[str | None]
2691 policy_queue_id: Mapped[int | None] = mapped_column(ForeignKey("policy_queue.id"))
2692 created: Mapped[datetime | None] = mapped_column(
2693 DateTime(timezone=True), server_default=func.now()
2694 )
2695 modified: Mapped[datetime | None] = mapped_column(
2696 DateTime(timezone=True), server_default=func.now()
2697 )
2698 changelog: Mapped[str | None]
2699 butautomaticupgrades: Mapped[bool] = mapped_column(default=False)
2700 signingkeys: Mapped[list[str] | None] = mapped_column(ARRAY(String()))
2701 announce: Mapped[list[str] | None] = mapped_column(ARRAY(String()))
2702 include_long_description: Mapped[bool] = mapped_column(default=False)
2703 overrideprocess: Mapped[bool] = mapped_column(default=False)
2704 overrideorigin: Mapped[str | None]
2705 allowcsset: Mapped[bool] = mapped_column(default=False)
2706 archive_id: Mapped[int] = mapped_column(ForeignKey("archive.id"))
2707 new_queue_id: Mapped[int | None] = mapped_column(ForeignKey("policy_queue.id"))
2708 close_bugs: Mapped[bool | None]
2709 mail_whitelist: Mapped[str | None]
2710 indices_compression: Mapped[list[str] | None] = mapped_column(
2711 ARRAY(String()), default=["xz"]
2712 )
2713 i18n_compression: Mapped[list[str] | None] = mapped_column(
2714 ARRAY(String()), default=["xz"]
2715 )
2716 release_suite: Mapped[str | None]
2717 debugsuite_id: Mapped[int | None] = mapped_column(ForeignKey("suite.id"))
2718 changelog_url: Mapped[str | None]
2719 accept_source_uploads: Mapped[bool | None] = mapped_column(default=True)
2720 accept_binary_uploads: Mapped[bool | None] = mapped_column(default=True)
2721 checksums: Mapped[list[str] | None] = mapped_column(
2722 ARRAY(String()), default=["sha256"]
2723 )
2724 last_changed: Mapped[datetime] = mapped_column(
2725 DateTime(timezone=False), server_default=func.now()
2726 )
2727 byhash: Mapped[bool | None] = mapped_column(default=True)
2728 separate_contents_architecture_all: Mapped[bool] = mapped_column(default=False)
2729 separate_packages_architecture_all: Mapped[bool] = mapped_column(default=False)
2730 merged_pdiffs: Mapped[bool] = mapped_column(default=True)
2731 stayofexecution: Mapped[timedelta] = mapped_column(default=timedelta(hours=0))
2733 policy_queue: Mapped["PolicyQueue | None"] = relationship(
2734 foreign_keys=[policy_queue_id]
2735 )
2736 new_queue: Mapped["PolicyQueue | None"] = relationship(foreign_keys=[new_queue_id])
2737 debug_suite: Mapped["Suite | None"] = relationship(remote_side=[suite_id])
2738 copy_queues: Mapped[list["BuildQueue"]] = relationship(
2739 secondary="suite_build_queue_copy"
2740 )
2741 srcformats: Mapped[list["SrcFormat"]] = relationship(
2742 secondary="suite_src_formats", back_populates="suites"
2743 )
2744 archive: Mapped[Archive] = relationship(back_populates="suites")
2745 acls: Mapped[set[ACL]] = relationship(secondary="suite_acl_map")
2746 components: Mapped[list["Component"]] = relationship(
2747 secondary="component_suite",
2748 back_populates="suites",
2749 order_by=lambda: Component.ordering,
2750 )
2751 architectures: Mapped[list[Architecture]] = relationship(
2752 secondary="suite_architectures", back_populates="suites"
2753 )
2754 binaries: DynamicMapped["DBBinary"] = relationship(
2755 secondary="bin_associations",
2756 back_populates="suites",
2757 )
2758 sources: DynamicMapped["DBSource"] = relationship(
2759 secondary="src_associations", back_populates="suites"
2760 )
2761 overrides: DynamicMapped["Override"] = relationship(back_populates="suite")
2763 def __init__(self, suite_name=None, version=None):
2764 self.suite_name = suite_name
2765 self.version = version
2767 @override
2768 def properties(self) -> list[str]:
2769 return [
2770 "suite_name",
2771 "version",
2772 "sources_count",
2773 "binaries_count",
2774 "overrides_count",
2775 ]
2777 @override
2778 def __eq__(self, val: object):
2779 if isinstance(val, str): 2779 ↛ 2780line 2779 didn't jump to line 2780 because the condition on line 2779 was never true
2780 warnings.warn(
2781 "comparison with a `str` is deprecated",
2782 DeprecationWarning,
2783 stacklevel=2,
2784 )
2785 return self.suite_name == val
2786 # This signals to use the normal comparison operator
2787 return NotImplemented
2789 @override
2790 def __ne__(self, val: object):
2791 if isinstance(val, str): 2791 ↛ 2792line 2791 didn't jump to line 2792 because the condition on line 2791 was never true
2792 warnings.warn(
2793 "comparison with a `str` is deprecated",
2794 DeprecationWarning,
2795 stacklevel=2,
2796 )
2797 return self.suite_name != val
2798 # This signals to use the normal comparison operator
2799 return NotImplemented
2801 __hash__ = ORMObject.__hash__
2803 def details(self) -> str:
2804 ret = []
2805 for disp, field in SUITE_FIELDS:
2806 val = getattr(self, field, None)
2807 if val is not None:
2808 ret.append("%s: %s" % (disp, val))
2810 return "\n".join(ret)
2812 def get_architectures(
2813 self, skipsrc: bool = False, skipall: bool = False
2814 ) -> list[Architecture]:
2815 """
2816 Returns list of Architecture objects
2818 :param skipsrc: Whether to skip returning the 'source' architecture entry
2819 :param skipall: Whether to skip returning the 'all' architecture entry
2820 :return: list of Architecture objects for the given name (may be empty)
2821 """
2823 session = object_session(self)
2824 assert session is not None
2825 q = select(Architecture).where(with_parent(self, Suite.architectures))
2826 if skipsrc:
2827 q = q.where(Architecture.arch_string != "source")
2828 if skipall:
2829 q = q.where(Architecture.arch_string != "all")
2830 return list(session.scalars(q.order_by(Architecture.arch_string)))
2832 def get_sources(self, source: str) -> "Select[tuple[DBSource]]":
2833 """
2834 Returns a select() statement for DBSource that is part of this suite.
2836 :param source: source package name
2837 :return: a select() of DBSource
2838 """
2840 return (
2841 select(DBSource)
2842 .where(DBSource.source == source)
2843 .where(with_parent(self, Suite.sources))
2844 )
2846 def get_overridesuite(self) -> "Suite":
2847 if self.overridesuite is None:
2848 return self
2849 session = object_session(self)
2850 assert session is not None
2851 q = select(Suite).where(Suite.suite_name == self.overridesuite)
2852 return session.execute(q).scalar_one()
2854 def update_last_changed(self) -> None:
2855 self.last_changed = sqlalchemy.func.now()
2857 @property
2858 def path(self) -> str:
2859 return os.path.join(self.archive.path, "dists", self.suite_name)
2861 @property
2862 def release_suite_output(self) -> str:
2863 if self.release_suite is not None: 2863 ↛ 2864line 2863 didn't jump to line 2864 because the condition on line 2863 was never true
2864 return self.release_suite
2865 return self.suite_name
2868__all__.append("Suite")
2871@session_wrapper
2872def get_suite(suite: str, session: "Session | None" = None) -> Suite | None:
2873 """
2874 Returns Suite object for given `suite` name.
2876 :param suite: The name of the suite
2877 :param session: Optional SQLA session object (a temporary one will be
2878 generated if not supplied)
2879 :return: Suite object for the requested suite name (None if not present)
2880 """
2882 assert session is not None
2884 # Start by looking for the dak internal name
2885 q = select(Suite).where(Suite.suite_name == suite)
2886 if (found := session.execute(q).scalar_one_or_none()) is not None:
2887 return found
2889 # Now try codename
2890 q = select(Suite).where(Suite.codename == suite)
2891 if (found := session.execute(q).scalar_one_or_none()) is not None: 2891 ↛ 2892line 2891 didn't jump to line 2892 because the condition on line 2891 was never true
2892 return found
2894 # Finally give release_suite a try
2895 q = select(Suite).where(Suite.release_suite == suite)
2896 return session.execute(q).scalar_one_or_none()
2899__all__.append("get_suite")
2902@session_wrapper
2903def get_suite_by_name(suite_name: str, session: "Session | None" = None) -> Suite:
2904 """
2905 Returns the Suite with the given name.
2907 Unlike get_suite() this does not fall back to codenames and an exception
2908 is raised if the suite does not exist.
2910 :param suite_name: the name of the suite
2911 :return: the Suite object for the given name
2912 """
2914 assert session is not None
2915 q = select(Suite).where(Suite.suite_name == suite_name)
2916 return session.execute(q).scalar_one()
2919__all__.append("get_suite_by_name")
2922class SuiteBuildQueueCopy(Base):
2923 __tablename__ = "suite_build_queue_copy"
2925 suite: Mapped[int] = mapped_column(ForeignKey("suite.id"), primary_key=True)
2926 build_queue_id: Mapped[int] = mapped_column(
2927 ForeignKey("build_queue.id"), primary_key=True
2928 )
2929 created: Mapped[datetime] = mapped_column(
2930 DateTime(timezone=True), server_default=func.now()
2931 )
2932 modified: Mapped[datetime] = mapped_column(
2933 DateTime(timezone=True), server_default=func.now()
2934 )
2937class SuiteSrcFormats(Base):
2938 __tablename__ = "suite_src_formats"
2940 suite: Mapped[int] = mapped_column(ForeignKey("suite.id"), primary_key=True)
2941 src_format: Mapped[int] = mapped_column(
2942 ForeignKey("src_format.id"), primary_key=True
2943 )
2944 created: Mapped[datetime] = mapped_column(
2945 DateTime(timezone=True), server_default=func.now()
2946 )
2947 modified: Mapped[datetime] = mapped_column(
2948 DateTime(timezone=True), server_default=func.now()
2949 )
2952class SuiteAclMap(Base):
2953 __tablename__ = "suite_acl_map"
2955 suite_id: Mapped[int] = mapped_column(
2956 ForeignKey("suite.id", ondelete="CASCADE"), primary_key=True
2957 )
2958 acl_id: Mapped[int] = mapped_column(ForeignKey("acl.id"), primary_key=True)
2961class SuiteArchitectures(Base):
2962 __tablename__ = "suite_architectures"
2964 suite: Mapped[int] = mapped_column(ForeignKey("suite.id"), primary_key=True)
2965 architecture: Mapped[int] = mapped_column(
2966 ForeignKey("architecture.id"), primary_key=True
2967 )
2968 created: Mapped[datetime] = mapped_column(
2969 DateTime(timezone=True), server_default=func.now()
2970 )
2971 modified: Mapped[datetime] = mapped_column(
2972 DateTime(timezone=True), server_default=func.now()
2973 )
2976################################################################################
2979@session_wrapper
2980def get_suite_architectures(
2981 suite_name: str,
2982 skipsrc: bool = False,
2983 skipall: bool = False,
2984 session: "Session | None" = None,
2985) -> list[Architecture]:
2986 """
2987 Returns list of Architecture objects for given `suite` name. The list is
2988 empty if `suite` does not exist.
2990 :param suite_name: Suite name to search for
2991 :param skipsrc: Whether to skip returning the 'source' architecture entry
2992 :param skipall: Whether to skip returning the 'all' architecture entry
2993 :param session: Optional SQL session object (a temporary one will be
2994 generated if not supplied)
2995 :return: list of Architecture objects for the given name (may be empty)
2996 """
2998 assert session is not None
2999 suite = get_suite(suite_name, session)
3000 if suite is None: 3000 ↛ 3001line 3000 didn't jump to line 3001 because the condition on line 3000 was never true
3001 return []
3002 return suite.get_architectures(skipsrc, skipall)
3005__all__.append("get_suite_architectures")
3007################################################################################
3010class Uid(ORMObject):
3011 __tablename__ = "uid"
3013 uid_id: Mapped[int] = mapped_column("id", primary_key=True)
3014 uid: Mapped[str] = mapped_column(unique=True)
3015 name: Mapped[str | None]
3016 created: Mapped[datetime] = mapped_column(
3017 DateTime(timezone=True), server_default=func.now()
3018 )
3019 modified: Mapped[datetime] = mapped_column(
3020 DateTime(timezone=True), server_default=func.now()
3021 )
3023 fingerprint: Mapped[list["Fingerprint"]] = relationship(back_populates="uid")
3025 def __init__(self, uid=None, name=None):
3026 self.uid = uid
3027 self.name = name
3029 @override
3030 def __eq__(self, val: object):
3031 if isinstance(val, str):
3032 warnings.warn(
3033 "comparison with a `str` is deprecated",
3034 DeprecationWarning,
3035 stacklevel=2,
3036 )
3037 return self.uid == val
3038 # This signals to use the normal comparison operator
3039 return NotImplemented
3041 @override
3042 def __ne__(self, val: object):
3043 if isinstance(val, str):
3044 warnings.warn(
3045 "comparison with a `str` is deprecated",
3046 DeprecationWarning,
3047 stacklevel=2,
3048 )
3049 return self.uid != val
3050 # This signals to use the normal comparison operator
3051 return NotImplemented
3053 __hash__ = ORMObject.__hash__
3055 @override
3056 def properties(self) -> list[str]:
3057 return ["uid", "name", "fingerprint"]
3060__all__.append("Uid")
3063@session_wrapper
3064def get_or_set_uid(uidname: str, session: "Session | None" = None) -> Uid:
3065 """
3066 Returns uid object for given uidname.
3068 If no matching uidname is found, a row is inserted.
3070 :param uidname: The uid to add
3071 :param session: Optional SQL session object (a temporary one will be
3072 generated if not supplied). If not passed, a commit will be performed at
3073 the end of the function, otherwise the caller is responsible for commiting.
3074 :return: the uid object for the given uidname
3075 """
3077 assert session is not None
3078 q = select(Uid).where(Uid.uid == uidname)
3080 ret = session.execute(q).scalar_one_or_none()
3081 if ret is None:
3082 uid = Uid()
3083 uid.uid = uidname
3084 session.add(uid)
3085 session.commit_or_flush() # type: ignore[attr-defined]
3086 ret = uid
3088 return ret
3091__all__.append("get_or_set_uid")
3094@session_wrapper
3095def get_uid_from_fingerprint(fpr: str, session: "Session | None" = None) -> Uid | None:
3096 assert session is not None
3097 q = select(Uid).join(Fingerprint).where(Fingerprint.fingerprint == fpr)
3099 return session.execute(q).scalar_one_or_none()
3102__all__.append("get_uid_from_fingerprint")
3104################################################################################
3107class MetadataKey(ORMObject):
3108 __tablename__ = "metadata_keys"
3110 key_id: Mapped[int] = mapped_column(primary_key=True)
3111 key: Mapped[str] = mapped_column(unique=True)
3112 ordering: Mapped[int] = mapped_column(default=0)
3114 def __init__(self, key=None):
3115 self.key = key
3117 @override
3118 def properties(self) -> list[str]:
3119 return ["key"]
3122__all__.append("MetadataKey")
3125@session_wrapper
3126def get_or_set_metadatakey(
3127 keyname: str, session: "Session | None" = None
3128) -> MetadataKey:
3129 """
3130 Returns MetadataKey object for given uidname.
3132 If no matching keyname is found, a row is inserted.
3134 :param keyname: The keyname to add
3135 :param session: Optional SQL session object (a temporary one will be
3136 generated if not supplied). If not passed, a commit will be performed at
3137 the end of the function, otherwise the caller is responsible for commiting.
3138 :return: the metadatakey object for the given keyname
3139 """
3141 assert session is not None
3142 q = select(MetadataKey).where(MetadataKey.key == keyname)
3144 ret = session.execute(q).scalar_one_or_none()
3145 if ret is None:
3146 ret = MetadataKey(keyname)
3147 session.add(ret)
3148 session.commit_or_flush() # type: ignore[attr-defined]
3150 return ret
3153__all__.append("get_or_set_metadatakey")
3155################################################################################
3158class BinaryMetadata(ORMObject):
3159 __tablename__ = "binaries_metadata"
3161 binary_id: Mapped[int] = mapped_column(
3162 "bin_id", ForeignKey("binaries.id", ondelete="CASCADE"), primary_key=True
3163 )
3164 key_id: Mapped[int] = mapped_column(
3165 ForeignKey("metadata_keys.key_id"), primary_key=True
3166 )
3167 value: Mapped[str]
3169 binary: Mapped["DBBinary"] = relationship(back_populates="key")
3170 key: Mapped[MetadataKey] = relationship()
3172 def __init__(self, key=None, value=None, binary=None):
3173 self.key = key
3174 self.value = value
3175 if binary is not None:
3176 self.binary = binary
3178 @override
3179 def properties(self) -> list[str]:
3180 return ["binary", "key", "value"]
3183__all__.append("BinaryMetadata")
3185################################################################################
3188class SourceMetadata(ORMObject):
3189 __tablename__ = "source_metadata"
3191 source_id: Mapped[int] = mapped_column(
3192 "src_id", ForeignKey("source.id", ondelete="CASCADE"), primary_key=True
3193 )
3194 key_id: Mapped[int] = mapped_column(
3195 ForeignKey("metadata_keys.key_id"), primary_key=True
3196 )
3197 value: Mapped[str]
3199 source: Mapped["DBSource"] = relationship(back_populates="key")
3200 key: Mapped[MetadataKey] = relationship()
3202 def __init__(self, key=None, value=None, source=None):
3203 self.key = key
3204 self.value = value
3205 if source is not None:
3206 self.source = source
3208 @override
3209 def properties(self) -> list[str]:
3210 return ["source", "key", "value"]
3213__all__.append("SourceMetadata")
3215################################################################################
3218class MetadataProxy:
3219 def __init__(self, session, query):
3220 self.session = session
3221 self.query = query
3223 def _get(self, key):
3224 metadata_key = self.session.scalars(
3225 select(MetadataKey).where(MetadataKey.key == key).limit(1)
3226 ).first()
3227 if metadata_key is None:
3228 return None
3229 metadata = self.session.scalars(
3230 self.query.filter_by(key=metadata_key).limit(1)
3231 ).first()
3232 return metadata
3234 def __contains__(self, key: str) -> bool:
3235 return self._get(key) is not None
3237 def __getitem__(self, key: str) -> str:
3238 metadata = self._get(key)
3239 if metadata is None:
3240 raise KeyError
3241 return metadata.value
3243 @overload
3244 def get(self, key: str, default: str) -> str: ... # noqa: E704 3244 ↛ exitline 3244 didn't return from function 'get' because
3246 @overload
3247 def get(self, key: str, default: None = None) -> str | None: ... # noqa: E704 3247 ↛ exitline 3247 didn't return from function 'get' because
3249 def get(self, key, default=None):
3250 try:
3251 return self[key]
3252 except KeyError:
3253 return default
3256################################################################################
3259class VersionCheck(ORMObject):
3260 __tablename__ = "version_check"
3262 suite_id: Mapped[int] = mapped_column(
3263 "suite", ForeignKey("suite.id"), primary_key=True
3264 )
3265 check: Mapped[str] = mapped_column(primary_key=True)
3266 reference_id: Mapped[int] = mapped_column(
3267 "reference", ForeignKey("suite.id"), primary_key=True
3268 )
3270 suite: Mapped["Suite"] = relationship(foreign_keys=[suite_id])
3271 reference: Mapped["Suite"] = relationship(
3272 foreign_keys=[reference_id], lazy="joined"
3273 )
3275 def __init__(self, *args, **kwargs):
3276 pass
3278 @override
3279 def properties(self) -> list[str]:
3280 return ["check"]
3283__all__.append("VersionCheck")
3286@session_wrapper
3287def get_version_checks(
3288 suite_name: str, check: str | None = None, session: "Session | None" = None
3289) -> list[VersionCheck]:
3290 assert session is not None
3291 suite = get_suite(suite_name, session)
3292 if not suite: 3292 ↛ 3295line 3292 didn't jump to line 3295 because the condition on line 3292 was never true
3293 # Make sure that what we return is iterable so that list comprehensions
3294 # involving this don't cause a traceback
3295 return []
3296 q = select(VersionCheck).filter_by(suite=suite)
3297 if check: 3297 ↛ 3299line 3297 didn't jump to line 3299 because the condition on line 3297 was always true
3298 q = q.filter_by(check=check)
3299 return list(session.scalars(q))
3302__all__.append("get_version_checks")
3304################################################################################
3307class ExternalSignatureRequests(Base):
3308 __tablename__ = "external_signature_requests"
3310 association_id: Mapped[int] = mapped_column(
3311 ForeignKey("bin_associations.id", ondelete="CASCADE"), primary_key=True
3312 )
3313 suite_id: Mapped[int] = mapped_column(
3314 ForeignKey("suite.id", ondelete="CASCADE"), primary_key=True
3315 )
3318################################################################################
3320package_list = Table(
3321 "package_list",
3322 Base.metadata,
3323 Column("package", String()),
3324 Column("version", DebVersion()),
3325 Column("source", String()),
3326 Column("source_version", DebVersion()),
3327 Column("suite", String()),
3328 Column("codename", String()),
3329 Column("archive", String()),
3330 Column("component", String()),
3331 Column("display_suite", String()),
3332 Column("architecture_is_source", Boolean()),
3333 Column("architecture", String()),
3334 Column("type", String()),
3335)
3337################################################################################
3340class DBConn:
3341 """
3342 database module init.
3343 """
3345 __shared_state: dict[str, Any] = {}
3347 def __init__(self, *, debug=False) -> None:
3348 self.__dict__ = self.__shared_state
3350 if not getattr(self, "initialised", False):
3351 self.initialised = True
3352 self.debug = debug
3353 self.__createconn()
3355 ## Connection functions
3356 def __createconn(self) -> None:
3357 from .config import Config
3359 cnf = Config()
3360 if "DB::Service" in cnf: 3360 ↛ 3361line 3360 didn't jump to line 3361 because the condition on line 3360 was never true
3361 connstr = "postgresql://service=%s" % cnf["DB::Service"]
3362 elif "DB::Host" in cnf:
3363 # TCP/IP
3364 connstr = "postgresql://%s" % cnf["DB::Host"]
3365 if "DB::Port" in cnf and cnf["DB::Port"] != "-1": 3365 ↛ 3366line 3365 didn't jump to line 3366 because the condition on line 3365 was never true
3366 connstr += ":%s" % cnf["DB::Port"]
3367 connstr += "/%s" % cnf["DB::Name"]
3368 else:
3369 # Unix Socket
3370 connstr = "postgresql:///%s" % cnf["DB::Name"]
3371 if "DB::Port" in cnf and cnf["DB::Port"] != "-1": 3371 ↛ 3372line 3371 didn't jump to line 3372 because the condition on line 3371 was never true
3372 connstr += "?port=%s" % cnf["DB::Port"]
3374 engine_args: dict[str, object] = {"echo": self.debug}
3375 if "DB::PoolSize" in cnf:
3376 engine_args["pool_size"] = int(cnf["DB::PoolSize"])
3377 if "DB::MaxOverflow" in cnf:
3378 engine_args["max_overflow"] = int(cnf["DB::MaxOverflow"])
3379 # we don't support non-utf-8 connections
3380 engine_args["client_encoding"] = "utf-8"
3382 # Monkey patch a new dialect in in order to support service= syntax
3383 import sqlalchemy.dialects.postgresql
3384 from sqlalchemy.dialects.postgresql.psycopg2 import PGDialect_psycopg2
3386 class PGDialect_psycopg2_dak(PGDialect_psycopg2):
3387 @override
3388 def create_connect_args(self, url):
3389 if str(url).startswith("postgresql://service="): 3389 ↛ 3391line 3389 didn't jump to line 3391 because the condition on line 3389 was never true
3390 # Eww
3391 servicename = str(url)[21:]
3392 return (["service=%s" % servicename], {})
3393 else:
3394 return PGDialect_psycopg2.create_connect_args(self, url)
3396 sqlalchemy.dialects.postgresql.base.dialect = PGDialect_psycopg2_dak # type: ignore[attr-defined]
3398 try:
3399 self.db_pg = create_engine(connstr, **engine_args)
3400 self.db_smaker = sessionmaker(
3401 bind=self.db_pg, autoflush=True, autocommit=False
3402 )
3403 except OperationalError as e:
3404 from . import utils
3406 utils.fubar("Cannot connect to database (%s)" % str(e))
3408 self.pid = os.getpid()
3410 def session(self, work_mem=0) -> "Session":
3411 """
3412 Returns a new session object. If a work_mem parameter is provided a new
3413 transaction is started and the work_mem parameter is set for this
3414 transaction. The work_mem parameter is measured in MB. A default value
3415 will be used if the parameter is not set.
3416 """
3417 # reinitialize DBConn in new processes
3418 if self.pid != os.getpid():
3419 self.__createconn()
3420 session = self.db_smaker()
3421 if work_mem > 0:
3422 session.execute(sql.text("SET LOCAL work_mem TO '%d MB'" % work_mem))
3423 return session
3426__all__.append("DBConn")