Skip to content

Commit

Permalink
Eliminate old-fashioned type annotations in storage service
Browse files Browse the repository at this point in the history
  • Loading branch information
nadove-ucsc authored and dsotirho-ucsc committed Feb 5, 2025
1 parent 8cabfae commit da88711
Showing 1 changed file with 10 additions and 11 deletions.
21 changes: 10 additions & 11 deletions src/azul/service/storage_service.py
Original file line number Diff line number Diff line change
Expand Up @@ -23,7 +23,6 @@
import time
from typing import (
IO,
Optional,
TYPE_CHECKING,
)
from urllib.parse import (
Expand Down Expand Up @@ -99,8 +98,8 @@ def get(self, object_key: str) -> bytes:
def put(self,
object_key: str,
data: bytes,
content_type: Optional[str] = None,
tagging: Optional[Tagging] = None,
content_type: str | None = None,
tagging: Tagging | None = None,
**kwargs):
self._s3.put_object(Bucket=self.bucket_name,
Key=object_key,
Expand All @@ -110,8 +109,8 @@ def put(self,

def create_multipart_upload(self,
object_key: str,
content_type: Optional[str] = None,
tagging: Optional[Tagging] = None) -> MultipartUpload:
content_type: str | None = None,
tagging: Tagging | None = None) -> MultipartUpload:
kwargs = self._object_creation_kwargs(content_type=content_type,
tagging=tagging)
return self._create_multipart_upload(object_key=object_key, **kwargs)
Expand Down Expand Up @@ -148,8 +147,8 @@ def complete_multipart_upload(self,
def upload(self,
file_path: str,
object_key: str,
content_type: Optional[str] = None,
tagging: Optional[Tagging] = None):
content_type: str | None = None,
tagging: Tagging | None = None):
self._s3.upload_file(Filename=file_path,
Bucket=self.bucket_name,
Key=object_key,
Expand All @@ -160,16 +159,16 @@ def upload(self,
self.put_object_tagging(object_key, tagging)

def _object_creation_kwargs(self, *,
content_type: Optional[str] = None,
tagging: Optional[Tagging] = None):
content_type: str | None = None,
tagging: Tagging | None = None):
kwargs = {}
if content_type is not None:
kwargs['ContentType'] = content_type
if tagging is not None:
kwargs['Tagging'] = urlencode(tagging)
return kwargs

def get_presigned_url(self, key: str, file_name: Optional[str] = None) -> str:
def get_presigned_url(self, key: str, file_name: str | None = None) -> str:
"""
Return a pre-signed URL to the given key.
Expand Down Expand Up @@ -264,7 +263,7 @@ def _time_until_object_expires(self,

@dataclass
class Part:
etag: Optional[str] # If ETag is defined, the content is already pushed to S3.
etag: str | None # If ETag is defined, the content is already pushed to S3.
part_number: int
content: bytes

Expand Down

0 comments on commit da88711

Please sign in to comment.