rpc.py 7.17 KB
Newer Older
1
2
3
from collections import defaultdict
from typing import List

4
5
from sqlalchemy import and_

6
7
import aurweb.config as config

8
from aurweb import db, models, util
9
from aurweb.models import dependency_type, relation_type
10

11
# Define dependency type mappings from ID to RPC-compatible keys.
12
DEP_TYPES = {
13
14
15
16
    dependency_type.DEPENDS_ID: "Depends",
    dependency_type.MAKEDEPENDS_ID: "MakeDepends",
    dependency_type.CHECKDEPENDS_ID: "CheckDepends",
    dependency_type.OPTDEPENDS_ID: "OptDepends"
17
18
}

19
# Define relationship type mappings from ID to RPC-compatible keys.
20
REL_TYPES = {
21
22
23
    relation_type.CONFLICTS_ID: "Conflicts",
    relation_type.PROVIDES_ID: "Provides",
    relation_type.REPLACES_ID: "Replaces"
24
25
26
}


27
28
class RPCError(Exception):
    pass
29
30


31
32
class RPC:
    """ RPC API handler class.
33

34
35
36
37
38
39
    There are various pieces to RPC's process, and encapsulating them
    inside of a class means that external users do not abuse the
    RPC implementation to achieve goals. We call type handlers
    by taking a reference to the callback named "_handle_{type}_type(...)",
    and if the handler does not exist, we return a not implemented
    error to the API user.
40

41
42
    EXPOSED_VERSIONS holds the set of versions that the API
    officially supports.
43

44
45
    EXPOSED_TYPES holds the set of types that the API officially
    supports.
46

47
    ALIASES holds an alias mapping of type -> type strings.
48

49
50
51
    We should focus on privatizing implementation helpers and
    focusing on performance in the code used.
    """
52

53
54
    # A set of RPC versions supported by this API.
    EXPOSED_VERSIONS = {5}
55

56
57
58
59
60
    # A set of RPC types supported by this API.
    EXPOSED_TYPES = {
        "info", "multiinfo",
        "search", "msearch",
        "suggest", "suggest-pkgbase"
61
62
    }

63
64
65
66
67
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
113
114
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
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
    # A mapping of aliases.
    ALIASES = {"info": "multiinfo"}

    def _verify_inputs(self, v: int, type: str, args: List[str] = []):
        if v is None:
            raise RPCError("Please specify an API version.")

        if v not in RPC.EXPOSED_VERSIONS:
            raise RPCError("Invalid version specified.")

        if type is None or not len(args):
            raise RPCError("No request type/data specified.")

        if type not in RPC.EXPOSED_TYPES:
            raise RPCError("Incorrect request type specified.")

        try:
            getattr(self, f"_handle_{type.replace('-', '_')}_type")
        except AttributeError:
            raise RPCError(f"Request type '{type}' is not yet implemented.")

    def _get_json_data(self, package: models.Package):
        """ Produce dictionary data of one Package that can be JSON-serialized.

        :param package: Package instance
        :returns: JSON-serializable dictionary
        """

        # Produce RPC API compatible Popularity: If zero, it's an integer
        # 0, otherwise, it's formatted to the 6th decimal place.
        pop = package.PackageBase.Popularity
        pop = 0 if not pop else float(util.number_format(pop, 6))

        snapshot_uri = config.get("options", "snapshot_uri")
        data = defaultdict(list)
        data.update({
            "ID": package.ID,
            "Name": package.Name,
            "PackageBaseID": package.PackageBaseID,
            "PackageBase": package.PackageBase.Name,
            # Maintainer should be set following this update if one exists.
            "Maintainer": None,
            "Version": package.Version,
            "Description": package.Description,
            "URL": package.URL,
            "URLPath": snapshot_uri % package.Name,
            "NumVotes": package.PackageBase.NumVotes,
            "Popularity": pop,
            "OutOfDate": package.PackageBase.OutOfDateTS,
            "FirstSubmitted": package.PackageBase.SubmittedTS,
            "LastModified": package.PackageBase.ModifiedTS,
            "License": [
                lic.License.Name for lic in package.package_licenses
            ],
            "Keywords": [
                keyword.Keyword for keyword in package.PackageBase.keywords
            ]
        })

        if package.PackageBase.Maintainer is not None:
            # We do have a maintainer: set the Maintainer key.
            data["Maintainer"] = package.PackageBase.Maintainer.Username

        # Walk through all related PackageDependencies and produce
        # the appropriate dict entries.
        if depends := package.package_dependencies:
            for dep in depends:
                if dep.DepTypeID in DEP_TYPES:
                    key = DEP_TYPES.get(dep.DepTypeID)

                    display = dep.DepName
                    if dep.DepCondition:
                        display += dep.DepCondition

                    data[key].append(display)

        # Walk through all related PackageRelations and produce
        # the appropriate dict entries.
        if relations := package.package_relations:
            for rel in relations:
                if rel.RelTypeID in REL_TYPES:
                    key = REL_TYPES.get(rel.RelTypeID)

                    display = rel.RelName
                    if rel.RelCondition:
                        display += rel.RelCondition

                    data[key].append(display)

        return data

    def _handle_multiinfo_type(self, args: List[str] = []):
        args = set(args)
        packages = db.query(models.Package).filter(
            models.Package.Name.in_(args))
        return [self._get_json_data(pkg) for pkg in packages]

    def _handle_suggest_pkgbase_type(self, args: List[str] = []):
        records = db.query(models.PackageBase).filter(
            and_(models.PackageBase.PackagerUID.isnot(None),
                 models.PackageBase.Name.like(f"%{args[0]}%"))
        ).order_by(models.PackageBase.Name.asc()).limit(20)
        return [record.Name for record in records]

    def handle(self, v: int = 0, type: str = None, args: List[str] = []):
        """ Request entrypoint. A router should pass v, type and args
        to this function and expect an output dictionary to be returned.

        :param v: RPC version argument
        :param type: RPC type argument
        :param args: Deciphered list of arguments based on arg/arg[] inputs
        """
        # Convert type aliased types.
        if type in RPC.ALIASES:
            type = RPC.ALIASES.get(type)

        # Prepare our output data dictionary with some basic keys.
        data = {"version": v, "type": type}

        # Run some verification on our given arguments.
        try:
            self._verify_inputs(v, type, args)
        except RPCError as exc:
            data.update({
                "results": [],
                "resultcount": 0,
                "type": "error",
                "error": str(exc)
            })
            return data

        # Get a handle to our callback and trap an RPCError with
        # an empty list of results based on callback's execution.
        callback = getattr(self, f"_handle_{type.replace('-', '_')}_type")
        results = callback(args)

        # These types are special: we produce a different kind of
        # successful JSON output: a list of results.
        if type in ("suggest", "suggest-pkgbase"):
            return results

        # Return JSON output.
        data.update({
            "resultcount": len(results),
            "results": results
        })
        return data