Database

models.py

Declaration of the DB model.

Package

Bases: Base

Table representing a Python Package registered in PyPi.

Source code in oblique/database/models.py

class Package(Base):
    """Table representing a Python Package registered in PyPi."""

    __tablename__ = "packages"

    id = Column(Integer, primary_key=True, index=True)
    last_updated = Column(DateTime)
    name = Column(String, unique=True, index=True)

    releases = relationship("Release", back_populates="package", passive_deletes=True)

Release

Bases: Base

Table representing a single release of a python package.

Source code in oblique/database/models.py

class Release(Base):
    """Table representing a single release of a python package."""

    __tablename__ = "releases"

    id = Column(Integer, primary_key=True, index=True)
    version = Column(String, nullable=False)
    date = Column(DateTime)
    is_yanked = Column(Boolean, default=False)
    package_id = Column(Integer, ForeignKey("packages.id", ondelete="CASCADE"))

    package = relationship("Package", back_populates="releases")

crud.py

CRUD functions to interact with the DB.

create_package(db, pkg_name)

CRUD function to create a new Package.

Parameters:

Name	Type	Description	Default
db	Session	DB Session.	required
pkg_name	str	Name of the package to create.	required

Returns:

Type	Description
Package	models.Package: Created Package.

Source code in oblique/database/crud.py

def create_package(db: Session, pkg_name: str) -> models.Package:
    """CRUD function to create a new Package.

    Args:
        db (Session): DB Session.
        pkg_name (str): Name of the package to create.

    Returns:
        models.Package: Created Package.
    """
    db_package = models.Package(name=pkg_name, last_updated=func.now())
    db.add(db_package)
    db.commit()
    db.refresh(db_package)
    return db_package

create_release(db, version, date, is_yanked, package_id)

CRUD function to create a new Release.

Parameters:

Name	Type	Description	Default
db	Session	DB Session.	required
version	str	Version name of this release.	required
date	datetime	Release date.	required
is_yanked	bool	If this release was yanked (True) or not (False).	required
package_id	int	ID of the package this release is associated with.	required

Returns:

Type	Description
Release	models.Release: Created Release.

Source code in oblique/database/crud.py

def create_release(db: Session, version: str, date: datetime, is_yanked: bool, package_id: int) -> models.Release:
    """CRUD function to create a new Release.

    Args:
        db (Session): DB Session.
        version (str): Version name of this release.
        date (datetime): Release date.
        is_yanked (bool): If this release was yanked (`True`) or not (`False`).
        package_id (int): ID of the package this release is associated with.

    Returns:
        models.Release: Created Release.
    """
    db_release = models.Release(version=version, date=date, is_yanked=is_yanked, package_id=package_id)
    db.add(db_release)
    db.commit()
    db.refresh(db_release)
    return db_release

create_releases(db, releases_data, package_id)

CRUD function to create several new Releases.

Parameters:

Name	Type	Description	Default
db	Session	DB Session.	required
releases_data	List[Tuple[str, datetime, bool]]	List of releases data. Each element is a tuple with the version name, the release date, and if this release is yanked or not.	required
package_id	int	ID of the package this release is associated with.	required

Source code in oblique/database/crud.py

def create_releases(db: Session, releases_data: List[Tuple[str, datetime, bool]], package_id: int):
    """CRUD function to create several new Releases.

    Args:
        db (Session): DB Session.
        releases_data (List[Tuple[str, datetime, bool]]): List of releases data.
            Each element is a tuple with the version name, the release date,
            and if this release is yanked or not.
        package_id (int): ID of the package this release is associated with.
    """
    for version, date, is_yanked in releases_data:
        create_release(db, version, date, is_yanked, package_id)

create_tables()

Util function to create the DB tables.

Warning, this function should be called only if the DB tables aren't already created.

Source code in oblique/database/crud.py

def create_tables():
    """Util function to create the DB tables.

    Warning, this function should be called only if the DB tables aren't
    already created.
    """
    models.Base.metadata.create_all(bind=engine)

get_latest_release_of(db, db_package)

CRUD function to retrieve the latest Release of a Package.

Parameters:

Name	Type	Description	Default
db	Session	DB Session.	required
db_package	Package	Package for which to get the last release date.	required

Returns:

Type	Description
Release	models.Release: Latest release for this package.

Source code in oblique/database/crud.py

def get_latest_release_of(db: Session, db_package: models.Package) -> models.Release:
    """CRUD function to retrieve the latest Release of a Package.

    Args:
        db (Session): DB Session.
        db_package (models.Package): Package for which to get the last release
            date.

    Returns:
        models.Release: Latest release for this package.
    """
    return (
        db.query(models.Release)
        .filter(models.Release.package_id == db_package.id)
        .order_by(models.Release.date.desc())
        .first()
    )

get_n_versions_of(db, db_package)

CRUD function to retrieve the number of releases of a Package.

Parameters:

Name	Type	Description	Default
db	Session	DB Session.	required
db_package	Package	Package for which to get the number of versions.	required

Returns:

Name	Type	Description
int	int	The number of versions for this package.

Source code in oblique/database/crud.py

def get_n_versions_of(db: Session, db_package: models.Package) -> int:
    """CRUD function to retrieve the number of releases of a Package.

    Args:
        db (Session): DB Session.
        db_package (models.Package): Package for which to get the number of
            versions.

    Returns:
        int: The number of versions for this package.
    """
    return db.query(models.Release).filter(models.Release.package_id == db_package.id).count()

get_n_versions_yanked_of(db, db_package)

CRUD function to retrieve the number of yanked releases of a Package.

Parameters:

Name	Type	Description	Default
db	Session	DB Session.	required
db_package	Package	Package for which to get the number of yanked versions.	required

Returns:

Name	Type	Description
int	int	The number of yanked versions for this package.

Source code in oblique/database/crud.py

def get_n_versions_yanked_of(db: Session, db_package: models.Package) -> int:
    """CRUD function to retrieve the number of yanked releases of a Package.

    Args:
        db (Session): DB Session.
        db_package (models.Package): Package for which to get the number of
            yanked versions.

    Returns:
        int: The number of yanked versions for this package.
    """
    return (
        db.query(models.Release)
        .filter(models.Release.package_id == db_package.id)
        .filter(models.Release.is_yanked.is_(True))
        .count()
    )

get_package_by_name(db, pkg_name)

CRUD function to retrieve a Package from its name.

Parameters:

Name	Type	Description	Default
db	Session	DB Session.	required
pkg_name	str	Name of the package to retrieve.	required

Returns:

Type	Description
Package	models.Package: Package with the given name.

Source code in oblique/database/crud.py

def get_package_by_name(db: Session, pkg_name: str) -> models.Package:
    """CRUD function to retrieve a Package from its name.

    Args:
        db (Session): DB Session.
        pkg_name (str): Name of the package to retrieve.

    Returns:
        models.Package: Package with the given name.
    """
    return db.query(models.Package).filter(models.Package.name == pkg_name).first()

update_package(db, db_package, releases_data)

CRUD function to update a Package (changing the last_updated attribute).

This will simply update the last_updated field, delete all associated releases, and recreate them from the given data.

Parameters:

Name	Type	Description	Default
db	Session	DB Session.	required
db_package	Package	Package to update.	required
releases_data	List[Tuple[str, datetime, bool]]	List of releases data. Each element is a tuple with the version name, the release date, and if this release is yanked or not.	required

Returns:

Type	Description
Package	models.Package: Package updated.

Source code in oblique/database/crud.py

def update_package(
    db: Session, db_package: models.Package, releases_data: List[Tuple[str, datetime, bool]]
) -> models.Package:
    """CRUD function to update a Package (changing the `last_updated`
    attribute).

    This will simply update the `last_updated` field, delete all associated
    releases, and recreate them from the given data.

    Args:
        db (Session): DB Session.
        db_package (models.Package): Package to update.
        releases_data (List[Tuple[str, datetime, bool]]): List of releases data.
            Each element is a tuple with the version name, the release date,
            and if this release is yanked or not.

    Returns:
        models.Package: Package updated.
    """
    # Update the element itself
    db_package.last_updated = func.now()
    db.commit()

    # Delete all of its previous releases
    db.query(models.Release).filter(models.Release.package_id == db_package.id).delete()

    # Recreate the releases from the fresh data
    create_releases(db, releases_data, db_package.id)

    db.refresh(db_package)
    return db_package