Changeset View
Changeset View
Standalone View
Standalone View
swh/lister/debian/models.py
| Show All 13 Lines | from sqlalchemy import ( | ||||
| Enum, | Enum, | ||||
| ForeignKey, | ForeignKey, | ||||
| Integer, | Integer, | ||||
| LargeBinary, | LargeBinary, | ||||
| String, | String, | ||||
| Table, | Table, | ||||
| UniqueConstraint, | UniqueConstraint, | ||||
| ) | ) | ||||
| from typing import Any, Mapping | |||||
| try: | try: | ||||
| from sqlalchemy import JSON | from sqlalchemy import JSON | ||||
| except ImportError: | except ImportError: | ||||
| # SQLAlchemy < 1.1 | # SQLAlchemy < 1.1 | ||||
| from sqlalchemy.dialects.postgresql import JSONB as JSON | from sqlalchemy.dialects.postgresql import JSONB as JSON | ||||
| from sqlalchemy.ext.declarative import declarative_base | from sqlalchemy.ext.declarative import declarative_base | ||||
| Show All 9 Lines | class Distribution(SQLBase): | ||||
| id = Column(Integer, primary_key=True) | id = Column(Integer, primary_key=True) | ||||
| name = Column(String, unique=True, nullable=False) | name = Column(String, unique=True, nullable=False) | ||||
| type = Column(Enum('deb', 'rpm', name='distribution_types'), | type = Column(Enum('deb', 'rpm', name='distribution_types'), | ||||
| nullable=False) | nullable=False) | ||||
| mirror_uri = Column(String, nullable=False) | mirror_uri = Column(String, nullable=False) | ||||
| areas = relationship('Area', back_populates='distribution') | areas = relationship('Area', back_populates='distribution') | ||||
| def origin_for_package(self, package_name, package_versions): | def origin_for_package( | ||||
| """Return the origin dictionary for the given package""" | self, package_name: str) -> str: | ||||
| return { | """Return the origin url for the given package | ||||
| 'type': self.type, | |||||
| 'url': '%s://%s/packages/%s' % ( | """ | ||||
| return '%s://%s/packages/%s' % ( | |||||
| self.type, self.name, package_name | self.type, self.name, package_name | ||||
| ), | ) | ||||
| } | |||||
| def __repr__(self): | def __repr__(self): | ||||
| return 'Distribution(%s (%s) on %s)' % ( | return 'Distribution(%s (%s) on %s)' % ( | ||||
| self.name, | self.name, | ||||
| self.type, | self.type, | ||||
| self.mirror_uri, | self.mirror_uri, | ||||
| ) | ) | ||||
| ▲ Show 20 Lines • Show All 115 Lines • ▼ Show 20 Lines | class DistributionSnapshot(SQLBase): | ||||
| date = Column(DateTime, nullable=False, index=True) | date = Column(DateTime, nullable=False, index=True) | ||||
| distribution_id = Column(Integer, | distribution_id = Column(Integer, | ||||
| ForeignKey('distribution.id'), | ForeignKey('distribution.id'), | ||||
| nullable=False) | nullable=False) | ||||
| distribution = relationship('Distribution') | distribution = relationship('Distribution') | ||||
| areas = relationship('AreaSnapshot', back_populates='snapshot') | areas = relationship('AreaSnapshot', back_populates='snapshot') | ||||
| def task_for_package(self, package_name, package_versions): | def task_for_package(self, package_name: str, | ||||
| """Return the task dictionary for the given list of package versions""" | package_versions: Mapping) -> Mapping[str, Any]: | ||||
| origin = self.distribution.origin_for_package( | """Return the task dictionary for the given list of package versions | ||||
| package_name, package_versions, | |||||
| ) | """ | ||||
| origin_url = self.distribution.origin_for_package(package_name) | |||||
| return { | return { | ||||
| 'policy': 'oneshot', | 'policy': 'oneshot', | ||||
| 'type': 'load-%s-package' % self.distribution.type, | 'type': 'load-%s-package' % self.distribution.type, | ||||
| 'next_run': datetime.datetime.now(tz=datetime.timezone.utc), | 'next_run': datetime.datetime.now(tz=datetime.timezone.utc), | ||||
| 'arguments': { | 'arguments': { | ||||
| 'args': [], | 'args': [], | ||||
| 'kwargs': { | 'kwargs': { | ||||
| 'url': origin, | 'url': origin_url, | ||||
| 'date': self.date.isoformat(), | 'date': self.date.isoformat(), | ||||
| 'packages': package_versions, | 'packages': package_versions, | ||||
| }, | }, | ||||
| } | }, | ||||
| 'retries_left': 3, | |||||
| } | } | ||||
| def get_packages(self): | def get_packages(self): | ||||
| packages = defaultdict(dict) | packages = defaultdict(dict) | ||||
| for area_snapshot in self.areas: | for area_snapshot in self.areas: | ||||
| area_name = area_snapshot.area.name | area_name = area_snapshot.area.name | ||||
| for package in area_snapshot.packages: | for package in area_snapshot.packages: | ||||
| ref_name = '%s/%s' % (area_name, package.version) | ref_name = '%s/%s' % (area_name, package.version) | ||||
| Show All 40 Lines | |||||