fix more typing

Author: khvn26

api/import_export/export.py

@@ -4,8 +4,12 @@
 import typing
 from collections.abc import Iterator
 from dataclasses import dataclass
+from typing import TYPE_CHECKING
 
 import boto3
+
+if TYPE_CHECKING:
+    from mypy_boto3_s3.client import S3Client
 from django.core import serializers
 from django.core.serializers.json import DjangoJSONEncoder
 from django.db.models import F, Model, Q
@@ -54,8 +58,8 @@
 
 
 class S3OrganisationExporter:
-    def __init__(self, s3_client: typing.Any = None) -> None:
-        self.s3_client = s3_client or boto3.client("s3")
+    def __init__(self, s3_client: "S3Client | None" = None) -> None:
+        self.s3_client: "S3Client" = s3_client or boto3.client("s3")
 
     def export_to_s3(
         self,

api/import_export/utils.py

@@ -2,6 +2,10 @@
 import logging
 import typing
 
+if typing.TYPE_CHECKING:
+    from mypy_boto3_s3.client import S3Client
+    from mypy_boto3_s3.type_defs import CompletedPartTypeDef
+
 logger = logging.getLogger(__name__)
 
 
@@ -19,15 +23,15 @@ class S3MultipartUploadWriter:
 
     def __init__(
         self,
-        s3_client: typing.Any,
+        s3_client: "S3Client",
         bucket_name: str,
         key: str,
     ) -> None:
         self._s3_client = s3_client
         self._bucket_name = bucket_name
         self._key = key
         self._buffer = io.BytesIO()
-        self._parts: list[dict[str, typing.Any]] = []
+        self._parts: list[CompletedPartTypeDef] = []
         self._part_number = 1
         self._upload_id: str | None = None
 
@@ -61,6 +65,7 @@ def __exit__(
         if self._buffer.tell() > 0:
             self._upload_part()
 
+        assert self._upload_id
         # Complete the multipart upload
         self._s3_client.complete_multipart_upload(
             Bucket=self._bucket_name,
@@ -76,6 +81,7 @@ def write(self, data: bytes) -> None:
             self._upload_part()
 
     def _upload_part(self) -> None:
+        assert self._upload_id
         self._buffer.seek(0)
         response = self._s3_client.upload_part(
             Bucket=self._bucket_name,

api/poetry.lock

@@ -227,6 +227,7 @@ requests-mock = "^1.11.0"
 django-extensions = "^3.2.3"
 pdbpp = "^0.10.3"
 mypy-boto3-dynamodb = "^1.33.0"
+mypy-boto3-s3 = "^1.36.0"
 pytest-structlog = "^1.1"
 pyfakefs = "^5.7.4"
 mypy = "^1.15.0"

api/pyproject.toml

@@ -61,6 +61,8 @@ def send_environment_update_message_for_environment(environment):  # type: ignor
 def stream_access_logs(
     timeout_seconds: int = 300,
 ) -> Generator[SSEAccessLogs, None, None]:
+    assert settings.AWS_SSE_LOGS_BUCKET_NAME
+
     gpg = gnupg.GPG(gnupghome=GNUPG_HOME)
     bucket = boto3.resource("s3").Bucket(settings.AWS_SSE_LOGS_BUCKET_NAME)
 

api/sse/sse_service.py

@@ -125,6 +125,7 @@ def test_stream_access_logs(mocker: MockerFixture, aws_credentials: None) -> Non
 
     # Next, let's create a bucket
     bucket_name = settings.AWS_SSE_LOGS_BUCKET_NAME
+    assert bucket_name
     s3_client = boto3.client("s3", region_name="eu-west-2")
     s3_client.create_bucket(
         Bucket=bucket_name,