Skip to content

core.py

File containing all the business logic, to be used by the API and the web-app.

Exceptions

Bases: Exception

Exception raised when the PyPi API doesn't respond or respond with an unhandled HTTP code.

Source code in oblique/core.py 
15
16
17
18
19
20
class PyPiAPIException(Exception):
    """Exception raised when the PyPi API doesn't respond or respond with an
    unhandled HTTP code.
    """

    pass

Bases: Exception

Exception raised when the given package name is not a package registered in PyPi index.

Source code in oblique/core.py 
23
24
25
26
27
28
class UnknownPackageException(Exception):
    """Exception raised when the given package name is not a package registered
    in PyPi index.
    """

    pass

Functions

Function that call the PyPi API and retrieve the releases data for a specific package name.

Parameters:

Name Type Description Default
pkg_name str

The package name for which we want to retrieve the data.

 required

Raises:

Type Description
PyPiAPIException

Exception raised if the PyPi API behaves unexpectedly.

Returns:

Type Description
List[Tuple[str, datetime, bool]]

List[Tuple[str, datetime, bool]]: Releases data for this package. It's a list, where each element represents a single release, which is a tuple with : * The version of the release * The release date * If this release was yanked (True) or not (False) Note that if this list is empty, it means the package does not exists in PyPi.
Source code in oblique/core.py 
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
def get_package_info_from_pypi(pkg_name: str) -> List[Tuple[str, datetime, bool]]:
    """Function that call the PyPi API and retrieve the releases data for a
    specific package name.

    Args:
        pkg_name (str): The package name for which we want to retrieve the data.

    Raises:
        PyPiAPIException: Exception raised if the PyPi API behaves unexpectedly.

    Returns:
        List[Tuple[str, datetime, bool]]: Releases data for this package. It's
            a list, where each element represents a single release, which is a
            tuple with :
            * The version of the release
            * The release date
            * If this release was yanked (`True`) or not (`False`)
            Note that if this list is empty, it means the package does not
            exists in PyPi.
    """
    r = requests.get(f"https://pypi.org/pypi/{pkg_name}/json")

    if r.status_code == 200:
        data = r.json()

        # Extract the data we need from the response
        return [
            (version, isoparse(info["upload_time"]), info["yanked"])
            for version, (info, *_) in data["releases"].items()
        ]
    elif r.status_code == 404:
        # Non-existing package : just return an empty list of releases
        return []
    else:
        raise PyPiAPIException("PyPi API unreachable")

Function that retrieve the statistics of a package stored in the DB.

Parameters:

Name Type Description Default
db Session

DB Session.

 required
db_package Package

The DB object corresponding to the package we want to extract statistics from.

 required
human_readable bool

If set to True, dates are returned in human-readable format (like 3 days ago for example). If set to False, dates are returned in ISO 8601. Defaults to True.

 True

Raises:

Type Description
UnknownPackageException

Exception raised if the package is not a package registered in PyPi index (has no releases).

Returns:

Type Description
Tuple[str, int, int]

Tuple[str, int, int]: The statistics for this package. This is a tuple with : * The release date in a human-readable format * The number of versions released * The number of versions yanked
Source code in oblique/core.py 
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
def get_stats_for(db: Session, db_package: models.Package, human_readable: bool = True) -> Tuple[str, int, int]:
    """Function that retrieve the statistics of a package stored in the DB.

    Args:
        db (Session): DB Session.
        db_package (models.Package): The DB object corresponding to the package
            we want to extract statistics from.
        human_readable (bool, optional): If set to `True`, dates are returned
            in human-readable format (like `3 days ago` for example). If set to
            `False`, dates are returned in ISO 8601. Defaults to `True`.

    Raises:
        UnknownPackageException: Exception raised if the package is not a
            package registered in PyPi index (has no releases).

    Returns:
        Tuple[str, int, int]: The statistics for this package. This is a tuple
            with :
            * The release date in a human-readable format
            * The number of versions released
            * The number of versions yanked
    """
    # From the DB, get our statistics
    last_release = crud.get_latest_release_of(db, db_package)
    n_versions = crud.get_n_versions_of(db, db_package)
    n_versions_yanked = crud.get_n_versions_yanked_of(db, db_package)

    # Handle the case where this package name wasn't released)
    if last_release is None:
        raise UnknownPackageException()

    # Format the last_release to be human-friendly
    if human_readable:
        td = datetime.utcnow() - last_release.date
        if td < timedelta(hours=24):
            h = max(td.seconds // 3600, 1)
            last_release = f"{h}h ago"
        elif td < timedelta(days=30):
            last_release = f"{td.days} day{'s' if td.days > 1 else ''} ago"
        else:
            last_release = f"{last_release.date:%d %b %Y}"
    else:
        last_release = last_release.date

    return last_release, n_versions, n_versions_yanked

Main function to retrieve informations about a PyPi package.

This function will first check if the informations is cached locally. If it's not cached locally, the data is retrieved from the PyPi API and cached locally. The cache is valid for 24h.

Parameters:

Name Type Description Default
db Session

DB Session.

 required
pkg_name str

Name of the package for which we want data.

 required
human_readable bool

If set to True, dates are returned in human-readable format (like 3 days ago for example). If set to False, dates are returned in ISO 8601. Defaults to True.

 True
force_refresh bool

If set to True, the local cache is ignored and the PyPi API is called. Note that it might be slower. Defaults to False.

 False

Returns:

Type Description
Tuple[str, int, int]

Tuple[str, int, int]: The statistics for the package. This is a tuple with : * The release date in a human-readable format * The number of versions released * The number of versions yanked
Source code in oblique/core.py 
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
def get_package_info(
    db: Session, pkg_name: str, human_readable: bool = True, force_refresh: bool = False
) -> Tuple[str, int, int]:
    """Main function to retrieve informations about a PyPi package.

    This function will first check if the informations is cached locally. If
    it's not cached locally, the data is retrieved from the PyPi API and cached
    locally.
    The cache is valid for 24h.

    Args:
        db (Session): DB Session.
        pkg_name (str): Name of the package for which we want data.
        human_readable (bool, optional): If set to `True`, dates are returned
            in human-readable format (like `3 days ago` for example). If set to
            `False`, dates are returned in ISO 8601. Defaults to `True`.
        force_refresh (bool, optional): If set to `True`, the local cache is
            ignored and the PyPi API is called. Note that it might be slower.
            Defaults to `False`.

    Returns:
        Tuple[str, int, int]: The statistics for the package. This is a tuple
            with :
            * The release date in a human-readable format
            * The number of versions released
            * The number of versions yanked
    """
    # Check the database to see if we already have that package's infos locally cached
    db_package = crud.get_package_by_name(db, pkg_name)

    if db_package is None or db_package.last_updated < datetime.utcnow() - CACHE_TTL or force_refresh:
        # This package is not cached locally, or the cache is stale
        # Call the PyPi API to retrieve its informations and update our local cache
        releases = get_package_info_from_pypi(pkg_name)

        if db_package is not None:
            db_package = crud.update_package(db, db_package, releases)
        else:
            db_package = crud.create_package(db, pkg_name)
            crud.create_releases(db, releases, db_package.id)

    # Retrieve the numbers we are interested in
    return get_stats_for(db, db_package, human_readable=human_readable)