packaging: add initial hints to storage services (#16709)

Signed-off-by: William Woodruff <[email protected]>
Co-authored-by: Dustin Ingram <[email protected]>

Author: woodruffw

warehouse/packaging/interfaces.py

@@ -26,25 +26,25 @@ def create_service(context, request):
         created for, passing a name for settings.
         """
 
-    def get(path):
+    def get(path: str):
         """
         Return a file like object that can be read to access the file located
         at the given path.
         """
 
-    def get_metadata(path):
+    def get_metadata(path: str):
         """
         Return a dictionary containing any user-created metadata associated
         with the file at a given path. Implementations may or may not store
         or provide such metadata.
         """
 
-    def get_checksum(path):
+    def get_checksum(path: str):
         """
         Return the md5 digest of the file at a given path as a lowercase string.
         """
 
-    def store(path, file_path, *, meta=None):
+    def store(path: str, file_path, *, meta=None):
         """
         Save the file located at file_path to the file storage at the location
         specified by path. An additional meta keyword argument may contain

warehouse/packaging/services.py

@@ -169,11 +169,11 @@ def remove_by_prefix(self, prefix):
 
 
 class GenericBlobStorage:
-    def __init__(self, bucket, *, prefix=None):
+    def __init__(self, bucket, *, prefix: str | None = None):
         self.bucket = bucket
         self.prefix = prefix
 
-    def _get_path(self, path):
+    def _get_path(self, path: str) -> str:
         # If we have a prefix, then prepend it to our path. This will let us
         # store items inside of a sub directory without exposing that to end
         # users.
@@ -184,7 +184,7 @@ def _get_path(self, path):
 
 
 class GenericB2BlobStorage(GenericBlobStorage):
-    def get(self, path):
+    def get(self, path: str):
         path = self._get_path(path)
         try:
             file_obj = io.BytesIO()
@@ -195,14 +195,14 @@ def get(self, path):
         except b2sdk.v2.exception.FileNotPresent:
             raise FileNotFoundError(f"No such key: {path!r}") from None
 
-    def get_metadata(self, path):
+    def get_metadata(self, path: str):
         path = self._get_path(path)
         try:
             return self.bucket.get_file_info_by_name(path).file_info
         except b2sdk.v2.exception.FileNotPresent:
             raise FileNotFoundError(f"No such key: {path!r}") from None
 
-    def get_checksum(self, path):
+    def get_checksum(self, path: str):
         path = self._get_path(path)
         try:
             return self.bucket.get_file_info_by_id(
@@ -211,7 +211,7 @@ def get_checksum(self, path):
         except b2sdk.v2.exception.FileNotPresent:
             raise FileNotFoundError(f"No such key: {path!r}") from None
 
-    def store(self, path, file_path, *, meta=None):
+    def store(self, path: str, file_path, *, meta=None):
         path = self._get_path(path)
         self.bucket.upload_local_file(
             local_file=file_path,
@@ -231,7 +231,7 @@ def create_service(cls, context, request):
 
 
 class GenericS3BlobStorage(GenericBlobStorage):
-    def get(self, path):
+    def get(self, path: str):
         # Note: this is not actually used to serve files, instead our CDN is
         # configured to connect directly to our storage bucket. See:
         # https://github.com/python/pypi-infra/blob/master/terraform/file-hosting/vcl/main.vcl
@@ -242,15 +242,15 @@ def get(self, path):
                 raise
             raise FileNotFoundError(f"No such key: {path!r}") from None
 
-    def get_metadata(self, path):
+    def get_metadata(self, path: str):
         try:
             return self.bucket.Object(self._get_path(path)).metadata
         except botocore.exceptions.ClientError as exc:
             if exc.response["Error"]["Code"] != "NoSuchKey":
                 raise
             raise FileNotFoundError(f"No such key: {path!r}") from None
 
-    def get_checksum(self, path):
+    def get_checksum(self, path: str):
         try:
             return (
                 self.bucket.Object(self._get_path(path)).e_tag.rstrip('"').lstrip('"')
@@ -261,7 +261,7 @@ def get_checksum(self, path):
                 raise
             raise FileNotFoundError(f"No such key: {path!r}") from None
 
-    def store(self, path, file_path, *, meta=None):
+    def store(self, path: str, file_path, *, meta=None):
         extra_args = {}
         if meta is not None:
             extra_args["Metadata"] = meta
@@ -327,24 +327,24 @@ def remove_by_prefix(self, prefix):
 
 
 class GenericGCSBlobStorage(GenericBlobStorage):
-    def get(self, path):
+    def get(self, path: str):
         # Note: this is not actually used in to serve files, instead our CDN is
         # configured to connect directly to our storage bucket. See:
         # https://github.com/python/pypi-infra/blob/master/terraform/file-hosting/vcl/main.vcl
         raise NotImplementedError
 
-    def get_metadata(self, path):
+    def get_metadata(self, path: str):
         raise NotImplementedError
 
-    def get_checksum(self, path):
+    def get_checksum(self, path: str):
         raise NotImplementedError
 
     @google.api_core.retry.Retry(
         predicate=google.api_core.retry.if_exception_type(
             google.api_core.exceptions.ServiceUnavailable
         )
     )
-    def store(self, path, file_path, *, meta=None):
+    def store(self, path: str, file_path, *, meta=None):
         path = self._get_path(path)
         blob = self.bucket.blob(path)
         if meta is not None: