Page Menu
Home
Software Heritage
Search
Configure Global Search
Log In
Files
F9346469
D8457.id30479.diff
No One
Temporary
Actions
View File
Edit File
Delete File
View Transforms
Subscribe
Mute Notifications
Award Token
Flag For Later
Size
7 KB
Subscribers
None
D8457.id30479.diff
View Options
diff --git a/swh/graphql/backends/archive.py b/swh/graphql/backends/archive.py
--- a/swh/graphql/backends/archive.py
+++ b/swh/graphql/backends/archive.py
@@ -4,61 +4,95 @@
# See top-level LICENSE file for more information
import os
-from typing import Any, Dict, Optional
+from typing import Any, Dict, Iterable, List, Optional
from swh.graphql import server
-from swh.model.model import Sha1Git
+from swh.model.model import (
+ Content,
+ DirectoryEntry,
+ Origin,
+ OriginVisit,
+ OriginVisitStatus,
+ Release,
+ Revision,
+ Sha1,
+ Sha1Git,
+)
from swh.model.swhids import ObjectType
+from swh.storage.interface import PagedResult, PartialBranches, StorageInterface
class Archive:
- def __init__(self):
- self.storage = server.get_storage()
+ def __init__(self) -> None:
+ self.storage: StorageInterface = server.get_storage()
- def get_origin(self, url):
- return self.storage.origin_get([url])[0]
+ def get_origin(self, url: str) -> Optional[Origin]:
+ return list(self.storage.origin_get(origins=[url]))[0]
- def get_origins(self, after=None, first=50):
+ def get_origins(
+ self, after: Optional[str] = None, first: int = 50
+ ) -> PagedResult[Origin]:
return self.storage.origin_list(page_token=after, limit=first)
- def get_origin_visits(self, origin_url, after=None, first=50):
- return self.storage.origin_visit_get(origin_url, page_token=after, limit=first)
+ def get_origin_visits(
+ self, origin_url: str, after: Optional[str] = None, first: int = 50
+ ) -> PagedResult[OriginVisit]:
+ return self.storage.origin_visit_get(
+ origin=origin_url, page_token=after, limit=first
+ )
- def get_origin_visit(self, origin_url, visit_id):
- return self.storage.origin_visit_get_by(origin_url, visit_id)
+ def get_origin_visit(self, origin_url: str, visit_id: int) -> Optional[OriginVisit]:
+ return self.storage.origin_visit_get_by(origin=origin_url, visit=visit_id)
- def get_origin_latest_visit(self, origin_url):
- return self.storage.origin_visit_get_latest(origin_url)
+ def get_origin_latest_visit(self, origin_url: str) -> Optional[OriginVisit]:
+ return self.storage.origin_visit_get_latest(origin=origin_url)
- def get_visit_status(self, origin_url, visit_id, after=None, first=50):
+ def get_visit_status(
+ self,
+ origin_url: str,
+ visit_id: int,
+ after: Optional[str] = None,
+ first: int = 50,
+ ) -> PagedResult[OriginVisitStatus]:
return self.storage.origin_visit_status_get(
- origin_url, visit_id, page_token=after, limit=first
+ origin=origin_url, visit=visit_id, page_token=after, limit=first
)
- def get_latest_visit_status(self, origin_url, visit_id):
- return self.storage.origin_visit_status_get_latest(origin_url, visit_id)
+ def get_latest_visit_status(
+ self, origin_url: str, visit_id: int
+ ) -> Optional[OriginVisitStatus]:
+ return self.storage.origin_visit_status_get_latest(
+ origin_url=origin_url, visit=visit_id
+ )
- def get_origin_snapshots(self, origin_url):
- return self.storage.origin_snapshot_get_all(origin_url)
+ def get_origin_snapshots(self, origin_url: str) -> List[Sha1Git]:
+ return self.storage.origin_snapshot_get_all(origin_url=origin_url)
def get_snapshot_branches(
- self, snapshot, after=b"", first=50, target_types=None, name_include=None
- ):
+ self,
+ snapshot: Sha1Git,
+ after: bytes = b"",
+ first: int = 50,
+ target_types: Optional[List[str]] = None,
+ name_include: Optional[bytes] = None,
+ ) -> Optional[PartialBranches]:
return self.storage.snapshot_get_branches(
- snapshot,
+ snapshot_id=snapshot,
branches_from=after,
branches_count=first,
target_types=target_types,
branch_name_include_substring=name_include,
)
- def get_revisions(self, revision_ids):
+ def get_revisions(self, revision_ids: List[Sha1Git]) -> List[Optional[Revision]]:
return self.storage.revision_get(revision_ids=revision_ids)
- def get_revision_log(self, revision_ids, after=None, first=50):
+ def get_revision_log(
+ self, revision_ids: List[Sha1Git], first: int = 50
+ ) -> Iterable[Optional[Dict[str, Any]]]:
return self.storage.revision_log(revisions=revision_ids, limit=first)
- def get_releases(self, release_ids):
+ def get_releases(self, release_ids: List[Sha1Git]) -> List[Optional[Release]]:
return self.storage.release_get(releases=release_ids)
def get_directory_entry_by_path(
@@ -69,12 +103,14 @@
directory=directory_id, paths=paths
)
- def get_directory_entries(self, directory_id, after=None, first=50):
+ def get_directory_entries(
+ self, directory_id: Sha1Git, after: Optional[bytes] = None, first: int = 50
+ ) -> Optional[PagedResult[DirectoryEntry]]:
return self.storage.directory_get_entries(
- directory_id, limit=first, page_token=after
+ directory_id=directory_id, limit=first, page_token=after
)
- def is_object_available(self, object_id: str, object_type: ObjectType) -> bool:
+ def is_object_available(self, object_id: bytes, object_type: ObjectType) -> bool:
mapping = {
ObjectType.CONTENT: self.storage.content_missing_per_sha1_git,
ObjectType.DIRECTORY: self.storage.directory_missing,
@@ -84,8 +120,8 @@
}
return not list(mapping[object_type]([object_id]))
- def get_contents(self, checksums: dict):
- return self.storage.content_find(checksums)
+ def get_contents(self, checksums: Dict[str, Any]) -> List[Content]:
+ return self.storage.content_find(content=checksums)
- def get_content_data(self, content_sha1):
- return self.storage.content_get_data(content_sha1)
+ def get_content_data(self, content_sha1: Sha1) -> Optional[bytes]:
+ return self.storage.content_get_data(content=content_sha1)
diff --git a/swh/graphql/backends/search.py b/swh/graphql/backends/search.py
--- a/swh/graphql/backends/search.py
+++ b/swh/graphql/backends/search.py
@@ -3,14 +3,20 @@
# License: GNU General Public License version 3, or any later version
# See top-level LICENSE file for more information
+from typing import Optional
+
from swh.graphql import server
+from swh.search.interface import MinimalOriginDict, SearchInterface
+from swh.storage.interface import PagedResult
class Search:
def __init__(self):
- self.search = server.get_search()
+ self.search: SearchInterface = server.get_search()
- def get_origins(self, query: str, after=None, first=50):
+ def get_origins(
+ self, query: str, after: Optional[str] = None, first: int = 50
+ ) -> PagedResult[MinimalOriginDict]:
return self.search.origin_search(
url_pattern=query,
page_token=after,
diff --git a/swh/graphql/server.py b/swh/graphql/server.py
--- a/swh/graphql/server.py
+++ b/swh/graphql/server.py
@@ -8,21 +8,23 @@
from swh.core import config
from swh.search import get_search as get_swh_search
+from swh.search.interface import SearchInterface
from swh.storage import get_storage as get_swh_storage
+from swh.storage.interface import StorageInterface
-graphql_cfg = None
-storage = None
-search = None
+graphql_cfg: Dict[str, Any] = {}
+storage: Optional[StorageInterface] = None
+search: Optional[SearchInterface] = None
-def get_storage():
+def get_storage() -> StorageInterface:
global storage
if not storage:
storage = get_swh_storage(**graphql_cfg["storage"])
return storage
-def get_search():
+def get_search() -> SearchInterface:
global search
if not search:
search = get_swh_search(**graphql_cfg["search"])
File Metadata
Details
Attached
Mime Type
text/plain
Expires
Thu, Jul 3, 4:05 PM (2 w, 2 d ago)
Storage Engine
blob
Storage Format
Raw Data
Storage Handle
3219423
Attached To
D8457: Add type hints to archive and search modules
Event Timeline
Log In to Comment