from __future__ import annotations import logging from typing import TYPE_CHECKING, Any import boto3 if TYPE_CHECKING: from mypy_boto3_s3 import S3Client as _S3Client from mypy_boto3_s3.type_defs import GetObjectOutputTypeDef, HeadObjectOutputTypeDef logger = logging.getLogger(__name__) class S3Client: def __init__(self, *, region_name: str, session: boto3.Session | None = None): self.region_name = region_name self._session = session self._client: _S3Client | None = None @property def session(self) -> boto3.Session: if self._session is None: self._session = boto3.Session() return self._session @property def client(self) -> _S3Client: if self._client is None: self._client = self.session.client("s3", region_name=self.region_name) return self._client def delete_object(self, bucket: str, key: str) -> None: self.client.delete_object(Bucket=bucket, Key=key) def head_object(self, bucket: str, key: str) -> HeadObjectOutputTypeDef: return self.client.head_object(Bucket=bucket, Key=key) def get_object(self, bucket: str, key: str) -> GetObjectOutputTypeDef: return self.client.get_object(Bucket=bucket, Key=key) def put_object( self, bucket: str, key: str, body: Any, content_type: str = "None", metadata: dict[str, Any] | None = None, ) -> str | None: self.client.put_object( Bucket=bucket, Key=key, Body=body, ContentType=content_type, Metadata=metadata or {}, )