laniakea.db package

Submodules

laniakea.db.archive module

class laniakea.db.archive.ArchiveArchitecture(name)

Bases: Base

A system architecture software can be compiled for. Usually associated with an :ArchiveSuite

id: Mapped[int]
name: Mapped[str]
pkgs_binary: Mapped[list[BinaryPackage]]
suites: Mapped[list[ArchiveSuite]]
summary: Mapped[str]
class laniakea.db.archive.ArchiveComponent(name)

Bases: Base

Information about an archive component within a suite.

id: Mapped[int]
is_nonfree()
is_primary()
name: Mapped[str]
parent_component: Mapped[ArchiveComponent]
parent_component_id: Mapped[int]
suites: Mapped[list[ArchiveSuite]]
summary: Mapped[str]
class laniakea.db.archive.ArchiveConfig(**kwargs)

Bases: Base

General archive configuration that applies to all repositories and suites.

archive_url: Mapped[str]
auto_debug_management: Mapped[bool]
extra_repo_root: Mapped[str]
id: Mapped[int]
primary_repo: Mapped[ArchiveRepository]
primary_repo_id: Mapped[int]
primary_repo_root: Mapped[str]
exception laniakea.db.archive.ArchiveError

Bases: Exception

Some issue occurred with the package archive.

class laniakea.db.archive.ArchiveFile(fname, repo=None)

Bases: Base

A file in the archive.

property absolute_repo_path: str | PathLike

Absolute path to the file in its repository.

NOTE: Files may also be in the queue directory of a repository, if they are pending review.

fname: Mapped[str]
id: Mapped[int]
make_url(urlbase)
md5sum: Mapped[str]
pkg_binary: Mapped[BinaryPackage]
pkgs_source: Mapped[list[SourcePackage]]
repo: Mapped[ArchiveRepository]
repo_id: Mapped[int]
sha1sum: Mapped[str]
sha256sum: Mapped[str]
sha512sum: Mapped[str]
size: Mapped[int]
time_created: Mapped[datetime]
class laniakea.db.archive.ArchiveQueueNewEntry(spkg, dest)

Bases: Base

Queue for package NEW processing.

comment: Mapped[str]
destination: Mapped[ArchiveSuite]
destination_id: Mapped[int]
id: Mapped[int]
package: Mapped[SourcePackage]
package_uuid: Mapped[UUID]
class laniakea.db.archive.ArchiveRepoSuiteSettings(repo, suite)

Bases: Base

Settings that are specific to a suite in a particular repository, but will not apply to the suite globally.

accept_uploads: Mapped[bool]
announce_emails: Mapped[list[str]]
auto_overrides: Mapped[bool]
but_automatic_upgrades: Mapped[bool]
changes_pending: Mapped[bool]
devel_target: Mapped[bool]
frozen: Mapped[bool]
id: Mapped[int]
manual_accept: Mapped[bool]
new_policy: Mapped[NewPolicy]
not_automatic: Mapped[bool]
phased_update_delay: Mapped[int]
repo: Mapped[ArchiveRepository]
repo_id: Mapped[int]
signingkeys: Mapped[list[str]]
suite: Mapped[ArchiveSuite]
suite_id: Mapped[int]
suite_summary: Mapped[str]
time_published: Mapped[datetime]
valid_time: Mapped[int]
class laniakea.db.archive.ArchiveRepository(name)

Bases: Base

A repository of packages.

debug_repo_for: Mapped[ArchiveRepository]
debug_repo_id: Mapped[int]
get_new_queue_dir()

Get the absolute path to the repository’s NEW queue directory

Return type:

str

get_new_queue_url()

Get the web URL to the repository’s NEW queue

Return type:

str

get_root_dir()

Get the absolute path to the repository’s root directory

Return type:

str

id: Mapped[int]
is_debug: Mapped[bool]
name: Mapped[str]
origin_name: Mapped[str | None]
suite_settings: Mapped[list[ArchiveRepoSuiteSettings]]
upload_suite_map: Mapped[dict]
uploaders: Mapped[list[ArchiveUploader]]
class laniakea.db.archive.ArchiveSection(name, title, summary=None)

Bases: Base

Known sections in the archive that packages are sorted into. See https://www.debian.org/doc/debian-policy/ch-archive.html#s-subsections for reference.

id: Mapped[int]
name: Mapped[str]
summary: Mapped[str]
title: Mapped[str]
class laniakea.db.archive.ArchiveSuite(name, alias=None)

Bases: Base

Information about suite in a distribution repository.

alias: Mapped[str]
architectures: Mapped[list[ArchiveArchitecture]]
components: Mapped[list[ArchiveComponent]]
dbgsym_policy: Mapped[DbgSymPolicy]
debug_suite: Mapped[ArchiveSuite | None]
debug_suite_for: Mapped[ArchiveSuite | None]
debug_suite_id: Mapped[int]
id: Mapped[int]
name: Mapped[str]
parents: Mapped[list[ArchiveSuite]]
pkgs_binary: Mapped[list[BinaryPackage]]
pkgs_source: Mapped[list[SourcePackage]]
property primary_architecture
repo_settings: Mapped[list[ArchiveRepoSuiteSettings]]
summary: Mapped[str]
version: Mapped[str]
class laniakea.db.archive.ArchiveUploader(email)

Bases: Base

Entities who are permitted to upload data to archive repositories.

alias: Mapped[str]
allow_binary_uploads: Mapped[bool]
allow_flatpak_uploads: Mapped[bool]
allow_source_uploads: Mapped[bool]
allowed_packages: Mapped[list[str]]
always_review: Mapped[bool]
email: Mapped[str]
id: Mapped[int]
is_human: Mapped[bool]
name: Mapped[str]
pgp_fingerprints: Mapped[list[str]]
repos: Mapped[list[ArchiveRepository]]
class laniakea.db.archive.ArchiveVersionMemory(**kwargs)

Bases: Base

Remember the highest version number for a source package that a repository has seen.

arch_name: Mapped[str]
highest_version: Mapped[DebVersion]
id: Mapped[int]
pkg_name: Mapped[str]
repo_suite: Mapped[ArchiveRepoSuiteSettings]
repo_suite_id: Mapped[int]
class laniakea.db.archive.BinaryPackage(name, version, repo=None)

Bases: Base

Data of a binary package.

architecture: Mapped[ArchiveArchitecture]
architecture_id: Mapped[int]
bin_file: Mapped[ArchiveFile]
bin_file_id: Mapped[int]
breaks: Mapped[list[str]]
build_ids: Mapped[list[str]]
built_using: Mapped[list[str]]
component: Mapped[ArchiveComponent]
component_id: Mapped[int]
conflicts: Mapped[list[str]]
contents: Mapped[list[str]]
deb_type: Mapped[DebType]
depends: Mapped[list[str]]
description: Mapped[str]
description_md5: Mapped[str]
property directory: str
enhances: Mapped[list[str]]
extra_data: Mapped[dict]
static generate_uuid(repo_name, name, version, arch_name)
homepage: Mapped[str]
maintainer: Mapped[str]
multi_arch: Mapped[str]
name: Mapped[str]
original_maintainer: Mapped[str]
phased_update_percentage: Mapped[int]
pre_depends: Mapped[list[str]]
provides: Mapped[list[str]]
recommends: Mapped[list[str]]
replaces: Mapped[list[str]]
repo: Mapped[ArchiveRepository]
repo_id: Mapped[int]
size_installed: Mapped[int]
source: Mapped[SourcePackage]
source_id: Mapped[UUID]
static_built_using: Mapped[list[str]]
suggests: Mapped[list[str]]
suites: Mapped[list[ArchiveSuite]]
summary: Mapped[str]
sw_cpts: Mapped[list[SoftwareComponent]]
time_added: Mapped[datetime]
time_deleted: Mapped[datetime]
time_published: Mapped[datetime]
update_uuid()
uuid: Mapped[UUID]
version: Mapped[str]
class laniakea.db.archive.ChangesUrgency(*values)

Bases: Enum

Urgency for how important it is to upgrade to a new package version from previous ones. https://www.debian.org/doc/debian-policy/ch-controlfields.html#urgency

CRITICAL = 4
EMERGENCY = 5
HIGH = 3
LOW = 1
MEDIUM = 2
UNKNOWN = 0
static from_string(s)

Convert the text representation into the enumerated type.

Return type:

ChangesUrgency

to_string()
class laniakea.db.archive.DbgSymPolicy(*values)

Bases: Enum

Policy for debug symbol handling for suites.

DEBUG_ALLOWED = 3
INVALID = 0
NO_DEBUG = 1
ONLY_DEBUG = 2
static from_string(s)

Convert the text representation into the enumerated type.

Return type:

DbgSymPolicy

to_string()
class laniakea.db.archive.DebType(*values)

Bases: IntEnum

Type of the Debian package.

DEB = 1
UDEB = 2
UNKNOWN = 0
static from_string(s)

Convert the text representation into the enumerated type.

Return type:

DebType

static to_string(e)
class laniakea.db.archive.NewPolicy(*values)

Bases: Enum

Policy for how new packages are processed.

ALWAYS_NEW = 2
DEFAULT = 1
INVALID = 0
NEVER_NEW = 3
static from_string(s)

Convert the text representation into the enumerated type.

Return type:

NewPolicy

to_string()
class laniakea.db.archive.PackageInfo(deb_type=DebType.DEB, name=None, version=None, component='main', section=None, essential=False, priority=PackagePriority.UNKNOWN, architectures=None)

Bases: object

Basic package information, used by :SourcePackage to refer to binary packages.

architectures: List[str] = None
component: str = 'main'
deb_type: DebType = 1
essential: bool = False
name: str = None
priority: PackagePriority = 0
section: str = None
version: str = None
class laniakea.db.archive.PackageOverride(pkgname, repo, suite)

Bases: Base

Overridable “archive organization” data of a binary package.

component: Mapped[ArchiveComponent]
component_id: Mapped[int]
essential: Mapped[bool]
id: Mapped[int]
pkg_name: Mapped[str]
priority: Mapped[PackagePriority]
repo: Mapped[ArchiveRepository]
repo_id: Mapped[int]
section: Mapped[ArchiveSection]
section_id: Mapped[int]
suite: Mapped[ArchiveSuite]
suite_id: Mapped[int]
class laniakea.db.archive.PackagePriority(*values)

Bases: IntEnum

Priority of a Debian package.

EXTRA = 5
IMPORTANT = 2
OPTIONAL = 4
REQUIRED = 1
STANDARD = 3
UNKNOWN = 0
static from_string(s)

Convert the text representation into the enumerated type.

Return type:

PackagePriority

static to_string(e)
class laniakea.db.archive.PackageType(*values)

Bases: Enum

Type of the package.

BINARY = 2
SOURCE = 1
UNKNOWN = 0
static from_string(s)

Convert the text representation into the enumerated type.

Return type:

PackageType

static to_string(e)
class laniakea.db.archive.SoftwareComponent(**kwargs)

Bases: Base

Description of a software component as described by the AppStream specification.

categories: Mapped[list[str]]
cid: Mapped[str]
property data: Dict[str, Any]
description: Mapped[str]
developer_name: Mapped[str]
flatpakref: Mapped[FlatpakRef]
flatpakref_uuid: Mapped[UUID]
gcid: Mapped[str]
icon_name: Mapped[str]
is_free: Mapped[bool]
kind: Mapped[int]
name: Mapped[str]
pkgs_binary: Mapped[list[BinaryPackage]]
project_license: Mapped[str]
summary: Mapped[str]
supports_touch: Mapped[bool]
update_uuid()

Update the unique identifier for this component.

uuid: Mapped[UUID]
static uuid_for_gcid(gcid)

Create an entity UUID from a component GCID

class laniakea.db.archive.SourcePackage(name, version, repo=None)

Bases: Base

Data of a source package.

architectures: Mapped[list[str]]
binaries: Mapped[list[BinaryPackage]]
build_conflicts: Mapped[list[str]]
build_conflicts_arch: Mapped[list[str]]
build_conflicts_indep: Mapped[list[str]]
build_depends: Mapped[list[str]]
build_depends_arch: Mapped[list[str]]
build_depends_indep: Mapped[list[str]]
changes_urgency: Mapped[ChangesUrgency]
component: Mapped[ArchiveComponent]
component_id: Mapped[int]
description: Mapped[str]
directory: Mapped[str]
property dsc_file: ArchiveFile | None
property expected_binaries: list[PackageInfo]
extra_data: Mapped[dict]
files: Mapped[list[ArchiveFile]]
format_version: Mapped[str]
static generate_source_uuid(repo_name, name)
static generate_uuid(repo_name, name, version)
get_metadata_dir(lconf=None)

Get the metadata storage location for this package.

homepage: Mapped[str]
maintainer: Mapped[str]
mark_remove()

Mark this source package for removal during next maintenance run.

name: Mapped[str]
original_maintainer: Mapped[str]
repo: Mapped[ArchiveRepository]
repo_id: Mapped[int]
section: Mapped[ArchiveSection]
section_id: Mapped[int]
source_uuid: Mapped[UUID]
standards_version: Mapped[str]
suites: Mapped[list[ArchiveSuite]]
summary: Mapped[str]
testsuite: Mapped[list[str]]
testsuite_triggers: Mapped[list[str]]
time_added: Mapped[datetime]
time_deleted: Mapped[datetime]
time_published: Mapped[datetime]
update_source_uuid()
update_uuid()
uploaders: Mapped[list[str]]
uuid: Mapped[UUID]
vcs_browser: Mapped[str]
vcs_git: Mapped[str]
version: Mapped[str]
laniakea.db.archive.package_version_compare(pkg1, pkg2)

Comparison function helper to compare package versions.

laniakea.db.base module

class laniakea.db.base.Base(**kwargs)

Bases: object

The base class of the class hierarchy.

When called, it accepts no arguments and returns a new featureless instance that has no instance attributes and cannot be given any.

metadata = MetaData()
registry = <sqlalchemy.orm.decl_api.registry object>
class laniakea.db.base.Database(lconf=None)

Bases: object

instance = None
class laniakea.db.base.DebVersion

Bases: UserDefinedType

bind_processor(dialect)

Return a conversion function for processing bind values.

Returns a callable which will receive a bind parameter value as the sole positional argument and will return a value to send to the DB-API.

If processing is not necessary, the method should return None.

Tip

This method is only called relative to a dialect specific type object, which is often private to a dialect in use and is not the same type object as the public facing one, which means it’s not feasible to subclass a types.TypeEngine class in order to provide an alternate _types.TypeEngine.bind_processor() method, unless subclassing the _types.UserDefinedType class explicitly.

To provide alternate behavior for _types.TypeEngine.bind_processor(), implement a _types.TypeDecorator class and provide an implementation of _types.TypeDecorator.process_bind_param().

See also

types_typedecorator

Parameters:

dialect – Dialect instance in use.

cache_ok: bool | None = True

Indicate if statements using this ExternalType are “safe to cache”.

The default value None will emit a warning and then not allow caching of a statement which includes this type. Set to False to disable statements using this type from being cached at all without a warning. When set to True, the object’s class and selected elements from its state will be used as part of the cache key. For example, using a TypeDecorator:

class MyType(TypeDecorator):
    impl = String

    cache_ok = True

    def __init__(self, choices):
        self.choices = tuple(choices)
        self.internal_only = True

The cache key for the above type would be equivalent to:

>>> MyType(["a", "b", "c"])._static_cache_key
(<class '__main__.MyType'>, ('choices', ('a', 'b', 'c')))

The caching scheme will extract attributes from the type that correspond to the names of parameters in the __init__() method. Above, the “choices” attribute becomes part of the cache key but “internal_only” does not, because there is no parameter named “internal_only”.

The requirements for cacheable elements is that they are hashable and also that they indicate the same SQL rendered for expressions using this type every time for a given cache value.

To accommodate for datatypes that refer to unhashable structures such as dictionaries, sets and lists, these objects can be made “cacheable” by assigning hashable structures to the attributes whose names correspond with the names of the arguments. For example, a datatype which accepts a dictionary of lookup values may publish this as a sorted series of tuples. Given a previously un-cacheable type as:

class LookupType(UserDefinedType):
    """a custom type that accepts a dictionary as a parameter.

    this is the non-cacheable version, as "self.lookup" is not
    hashable.

    """

    def __init__(self, lookup):
        self.lookup = lookup

    def get_col_spec(self, **kw):
        return "VARCHAR(255)"

    def bind_processor(self, dialect): ...  # works with "self.lookup" ...

Where “lookup” is a dictionary. The type will not be able to generate a cache key:

>>> type_ = LookupType({"a": 10, "b": 20})
>>> type_._static_cache_key
<stdin>:1: SAWarning: UserDefinedType LookupType({'a': 10, 'b': 20}) will not
produce a cache key because the ``cache_ok`` flag is not set to True.
Set this flag to True if this type object's state is safe to use
in a cache key, or False to disable this warning.
symbol('no_cache')

If we did set up such a cache key, it wouldn’t be usable. We would get a tuple structure that contains a dictionary inside of it, which cannot itself be used as a key in a “cache dictionary” such as SQLAlchemy’s statement cache, since Python dictionaries aren’t hashable:

>>> # set cache_ok = True
>>> type_.cache_ok = True

>>> # this is the cache key it would generate
>>> key = type_._static_cache_key
>>> key
(<class '__main__.LookupType'>, ('lookup', {'a': 10, 'b': 20}))

>>> # however this key is not hashable, will fail when used with
>>> # SQLAlchemy statement cache
>>> some_cache = {key: "some sql value"}
Traceback (most recent call last): File "<stdin>", line 1,
in <module> TypeError: unhashable type: 'dict'

The type may be made cacheable by assigning a sorted tuple of tuples to the “.lookup” attribute:

class LookupType(UserDefinedType):
    """a custom type that accepts a dictionary as a parameter.

    The dictionary is stored both as itself in a private variable,
    and published in a public variable as a sorted tuple of tuples,
    which is hashable and will also return the same value for any
    two equivalent dictionaries.  Note it assumes the keys and
    values of the dictionary are themselves hashable.

    """

    cache_ok = True

    def __init__(self, lookup):
        self._lookup = lookup

        # assume keys/values of "lookup" are hashable; otherwise
        # they would also need to be converted in some way here
        self.lookup = tuple((key, lookup[key]) for key in sorted(lookup))

    def get_col_spec(self, **kw):
        return "VARCHAR(255)"

    def bind_processor(self, dialect): ...  # works with "self._lookup" ...

Where above, the cache key for LookupType({"a": 10, "b": 20}) will be:

>>> LookupType({"a": 10, "b": 20})._static_cache_key
(<class '__main__.LookupType'>, ('lookup', (('a', 10), ('b', 20))))

Added in version 1.4.14: - added the cache_ok flag to allow some configurability of caching for TypeDecorator classes.

Added in version 1.4.28: - added the ExternalType mixin which generalizes the cache_ok flag to both the TypeDecorator and UserDefinedType classes.

See also

sql_caching

get_col_spec(**kw)
result_processor(dialect, coltype)

Return a conversion function for processing result row values.

Returns a callable which will receive a result row column value as the sole positional argument and will return a value to return to the user.

If processing is not necessary, the method should return None.

Tip

This method is only called relative to a dialect specific type object, which is often private to a dialect in use and is not the same type object as the public facing one, which means it’s not feasible to subclass a types.TypeEngine class in order to provide an alternate _types.TypeEngine.result_processor() method, unless subclassing the _types.UserDefinedType class explicitly.

To provide alternate behavior for _types.TypeEngine.result_processor(), implement a _types.TypeDecorator class and provide an implementation of _types.TypeDecorator.process_result_value().

See also

types_typedecorator

Parameters:

  • dialect – Dialect instance in use.

  • coltype – DBAPI coltype argument received in cursor.description.

class laniakea.db.base.UUID(as_uuid=True)

Bases: Uuid, NativeForEmulated

Represent the SQL UUID type.

This is the SQL-native form of the _types.Uuid database agnostic datatype, and is backwards compatible with the previous PostgreSQL-only version of UUID.

The _sqltypes.UUID datatype only works on databases that have a SQL datatype named UUID. It will not function for backends which don’t have this exact-named type, including SQL Server. For backend-agnostic UUID values with native support, including for SQL Server’s UNIQUEIDENTIFIER datatype, use the _sqltypes.Uuid datatype.

Added in version 2.0.

See also

_sqltypes.Uuid

classmethod adapt_emulated_to_native(impl, **kw)

Given an impl, adapt this type’s class to the impl assuming “native”.

The impl will be an Emulated class but not a NativeForEmulated.

e.g.: postgresql.ENUM produces a type given an Enum instance.

laniakea.db.base.create_tsvector(*args)
laniakea.db.base.print_query(query, literals=True)

Print a SQLAlchemy query with literals inserted and adjusted for the PostgreSQL dialect.

laniakea.db.base.session_scope(*, autoflush=True)

Provide a transactional scope around a series of operations.

laniakea.db.core module

class laniakea.db.core.ConfigEntry(mod, identifier, value=None)

Bases: Base

A generic, multi-purpose configuration entry.

id: Mapped[str]
set_value(mod, key, value)
value: Mapped[dict]
class laniakea.db.core.LkModule

Bases: object

String identifiers of Laniakea modules.

ADMINCLI = 'admin-cli'
ARCHIVE = 'archive'
ARIADNE = 'ariadne'
BASE = 'core'
DEBCHECK = 'debcheck'
ISOTOPE = 'isotope'
KEYTOOL = 'keytool'
LIGHTHOUSE = 'lighthouse'
PLANTER = 'planter'
RUBICON = 'rubicon'
SPEARS = 'spears'
SYNCHROTRON = 'synchrotron'
TESTSUITE = 'test'
UNKNOWN = ''
WEBDASH = 'webdash'
WEBSWVIEW = 'webswview'
laniakea.db.core.config_get_distro_tag()

Retrieve version tag for this distribution (“pureos”, “tanglu”, …) - will usually be part of a package version, e.g. “1.0-0tanglu1”

laniakea.db.core.config_get_project_name()

Get the name of the distribution or project (“Tanglu”, “PureOS”, …)

laniakea.db.core.config_get_value(mod, key)

Get a value from the configuration store.

laniakea.db.core.config_set_distro_tag(value)

Set version tag for this distribution (“pureos”, “tanglu”, …) - will usually be part of a package version, e.g. “1.0-0tanglu1”

laniakea.db.core.config_set_project_name(value)

Set the name of the distribution or project (“Tanglu”, “PureOS”, …)

laniakea.db.core.config_set_value(mod, key, value)

Set a value in the configuration store

laniakea.db.debcheck module

class laniakea.db.debcheck.DebcheckIssue(**kwargs)

Bases: Base

Data for a package migration excuse, as emitted by Britney

architectures: Mapped[list[str]]
property conflicts
static generate_uuid(issue, repo, suite)

Issue entities have a UUID based on a set of data, this function generates the UUID.

property missing
package_name: Mapped[str]
package_type: Mapped[PackageType]
package_version: Mapped[DebVersion]
repo: Mapped[ArchiveRepository]
repo_id: Mapped[int]
suite: Mapped[ArchiveSuite]
suite_id: Mapped[int]
time: Mapped[datetime]
update_uuid(repo=None, suite=None)
uuid: Mapped[UUID]
class laniakea.db.debcheck.PackageConflict(*, only=None, exclude=(), many=None, load_only=(), dump_only=(), partial=None, unknown=None)

Bases: Schema

Information about a conflict between packages.

class Meta

Bases: object

unknown = 'exclude'
opts: Any = <marshmallow.schema.SchemaOpts object>
class laniakea.db.debcheck.PackageIssue(*, only=None, exclude=(), many=None, load_only=(), dump_only=(), partial=None, unknown=None)

Bases: Schema

Information about the package issue reason.

class Meta

Bases: object

unknown = 'exclude'
architectures: list[str]
depends: str | None
opts: Any = <marshmallow.schema.SchemaOpts object>
package_name: str
package_type: PackageType
package_version: str
unsat_conflict: str
unsat_dependency: str

laniakea.db.flatpak module

class laniakea.db.flatpak.FlatpakRef(**kwargs)

Bases: Base

Flatpak object/app/runtime reference.

architecture: Mapped[ArchiveArchitecture]
architecture_id: Mapped[int]
branch: Mapped[str]
commit: Mapped[bytes]
kind: Mapped[FlatpakRefKind]
name: Mapped[str]
repo: Mapped[FlatpakRepository]
repo_id: Mapped[int]
uuid: Mapped[UUID]
version: Mapped[DebVersion]
class laniakea.db.flatpak.FlatpakRefKind(*values)

Bases: Enum

Kind of a Flatpak Ref.

APP = 2
RUNTIME = 3
UNKNOWN = 1
to_string()
class laniakea.db.flatpak.FlatpakRepository(name)

Bases: Base

A Flatpak repository.

allowed_branches: Mapped[list[str]]
collection_id: Mapped[str]
comment: Mapped[str]
default_branch: Mapped[str]
description: Mapped[str]
gpg_key_id: Mapped[str]
id: Mapped[int]
name: Mapped[str]
title: Mapped[str]
url_homepage: Mapped[str]
url_icon: Mapped[str]

laniakea.db.isotope module

class laniakea.db.isotope.ImageBuildRecipe(**kwargs)

Bases: Base

Instructions on how to do an automatic ISO image build.

architectures: Mapped[list[str]]
distribution: Mapped[str]
environment: Mapped[str]
format: Mapped[ImageFormat]
git_url: Mapped[str]
host_architecture: Mapped[str]
name: Mapped[str]
result_move_to: Mapped[str]
retain_images_n: Mapped[int]
style: Mapped[str]
suite: Mapped[str]
uuid: Mapped[UUID]
class laniakea.db.isotope.ImageFormat(*values)

Bases: IntEnum

Kind of the image to build.

IMG = 2
ISO = 1
UNKNOWN = 0

laniakea.db.jobs module

class laniakea.db.jobs.Job(**kwargs)

Bases: Base

A task to be performed (e.g. by a Spark worker)

architecture: Mapped[str]
data: Mapped[dict]
has_result()
is_failed()
is_taken()
kind: Mapped[str]
latest_log_excerpt: Mapped[str]
module: Mapped[str]
priority: Mapped[int]
result: Mapped[JobResult]
status: Mapped[JobStatus]
suite: Mapped[ArchiveSuite]
suite_id: Mapped[int]
time_assigned: Mapped[datetime]
time_created: Mapped[datetime]
time_finished: Mapped[datetime]
trigger: Mapped[UUID]
uuid: Mapped[UUID]
version: Mapped[DebVersion]
worker: Mapped[UUID]
class laniakea.db.jobs.JobKind

Bases: object

The different job kind identifier strings used by the different Laniakea modules which can enqueue jobs.

OS_IMAGE_BUILD = 'os-image-build'
PACKAGE_BUILD = 'package-build'
class laniakea.db.jobs.JobResult(*values)

Bases: IntEnum

Result of a job.

FAILURE = 5
FAILURE_DEPENDENCY = 3
FAILURE_PENDING = 4
SUCCESS = 2
SUCCESS_PENDING = 1
UNKNOWN = 0
class laniakea.db.jobs.JobStatus(*values)

Bases: IntEnum

State of a job.

DEPWAIT = 2
DONE = 5
RUNNING = 4
SCHEDULED = 3
STARVING = 7
TERMINATED = 6
UNKNOWN = 0
WAITING = 1

laniakea.db.spears module

class laniakea.db.spears.SpearsExcuse(**kwargs)

Bases: Base

Data for a package migration excuse, as emitted by Britney

age_current: Mapped[int]
age_required: Mapped[int]
blocked_by: Mapped[list[str]]
get_manual_block_hints()
get_old_binaries()
is_candidate: Mapped[bool]
log_excerpt: Mapped[str]
maintainer: Mapped[str]
make_idname()
manual_block: Mapped[dict]
migrate_after: Mapped[list[str]]
migration_id: Mapped[int]
migration_task: Mapped[SpearsMigrationTask]
missing_archs_primary: Mapped[list[str]]
missing_archs_secondary: Mapped[list[str]]
old_binaries: Mapped[dict]
other: Mapped[list[str]]
set_old_binaries(obins)
source_package: Mapped[SourcePackage]
source_package_id: Mapped[UUID]
time_created: Mapped[datetime]
uuid: Mapped[UUID]
version_new: Mapped[str]
version_old: Mapped[str]
class laniakea.db.spears.SpearsHint(**kwargs)

Bases: Base

User-defined hints for Britney.

hint: Mapped[str]
migration_id: Mapped[int]
migration_task: Mapped[SpearsMigrationTask]
reason: Mapped[str]
time: Mapped[datetime]
user: Mapped[str]
uuid: Mapped[UUID]
class laniakea.db.spears.SpearsMigrationTask(**kwargs)

Bases: Base

Description of a migration task from one or multiple suites to a target.

delays: Mapped[dict]
id: Mapped[int]
make_migration_shortname()

get a short name for this migration that can be used in file paths.

Return type:

str

make_migration_unique_name()

Get a unique identifier for this migration task

repo: Mapped[ArchiveRepository]
repo_id: Mapped[int]
source_suites: Mapped[list[ArchiveSuite]]
property source_suites_str: str

Get a string identifying the source suites packages are migrated from.

target_suite: Mapped[ArchiveSuite]
target_suite_id: Mapped[int]
class laniakea.db.spears.SpearsOldBinaries

Bases: object

List of old binaries of a specific version that a package has left behind.

binaries: list[Any] = []
pkg_version: str = ''

laniakea.db.synchrotron module

class laniakea.db.synchrotron.SyncBlacklistEntry(**kwargs)

Bases: Base

Synchrotron blacklist

config: Mapped[SynchrotronConfig]
config_id: Mapped[int]
pkgname: Mapped[str]
reason: Mapped[str]
time_created: Mapped[datetime]
user: Mapped[str]
uuid: Mapped[UUID]
class laniakea.db.synchrotron.SynchrotronConfig(**kwargs)

Bases: Base

Configuration for automatic synchrotron tasks.

auto_cruft_remove: Mapped[bool]
destination_suite: Mapped[ArchiveSuite]
destination_suite_id: Mapped[int]
id: Mapped[int]
repo: Mapped[ArchiveRepository]
repo_id: Mapped[int]
source: Mapped[SynchrotronSource]
source_id: Mapped[int]
sync_auto_enabled: Mapped[bool]
sync_binaries: Mapped[bool]
sync_enabled: Mapped[bool]
class laniakea.db.synchrotron.SynchrotronIssue(**kwargs)

Bases: Base

Hints about why packages are not synchronized with a source distribution/suite.

config: Mapped[SynchrotronConfig]
config_id: Mapped[int]
details: Mapped[str]
kind: Mapped[SynchrotronIssueKind]
package_name: Mapped[str]
source_suite: Mapped[str]
source_version: Mapped[DebVersion]
target_suite: Mapped[str]
target_version: Mapped[DebVersion]
time_created: Mapped[datetime]
uuid: Mapped[UUID]
class laniakea.db.synchrotron.SynchrotronIssueKind(*values)

Bases: IntEnum

Kind of a Synchrotron issue.

MAYBE_CRUFT = 3
MERGE_REQUIRED = 2
NONE = 1
REMOVAL_FAILED = 5
SYNC_FAILED = 4
UNKNOWN = 0
to_string()
class laniakea.db.synchrotron.SynchrotronSource(**kwargs)

Bases: Base

Definition of a foreign suite to sync packages from.

architectures: Mapped[list[str]]
components: Mapped[list[str]]
id: Mapped[int]
os_name: Mapped[str]
repo_url: Mapped[str]
suite_name: Mapped[str]

laniakea.db.workers module

class laniakea.db.workers.SparkWorker(**kwargs)

Bases: Base

An external machine/service that takes tasks from a Lighthouse server.

accepts: Mapped[list[str]]
architectures: Mapped[list[str]]
data: Mapped[dict]
enabled: Mapped[bool]
last_ping: Mapped[datetime]
name: Mapped[str]
owner: Mapped[str]
status: Mapped[WorkerStatus]
time_created: Mapped[datetime]
uuid: Mapped[UUID]
class laniakea.db.workers.WorkerStatus(*values)

Bases: IntEnum

State this worker is in.

ACTIVE = 1
DEAD = 4
IDLE = 2
MISSING = 3
UNKNOWN = 0

Module contents