I tried to use aiobotocore for async work with Ceph, but i have no access to real bucket, then i need to mock s3 bucket to create a fake one
I’m trying to use moto but it anyway want to connect a real bucket
<code>
@moto.mock_aws
async def _load_asl_by_type(self, file: File, typ: ContentType) -> AsyncIterable[Tuple[str, Content]]:
session = get_session()
async with session.create_client('s3', region_name='us-east-1', aws_secret_access_key='moto',
aws_access_key_id='moto') as s3:
bucket_name = 'y-bucket'
await s3.create_bucket(Bucket=bucket_name)
await s3.put_object(Bucket=bucket_name, Key=file.path)
try:
async with AsyncFileIO(io=await s3.get_object(Bucket=bucket_name, Key=file.path)) as f:
file.in_progress()
collection = json.load(f)
for name, data in collection.items():
def _spec_load() -> Any:
_logger.info(f"Loading {typ.__qualname__} from {file.path}: {name}")
return spec.load(typ, data, key=name, options=self._options)
obj = await asyncio.get_running_loop().run_in_executor(None, _spec_load)
file.loaded()
yield name, obj
except Exception as error:
_logger.error(f"Failed to load file {file.path}: {error}")
file.failed()
raise error
</code>
<code>
@moto.mock_aws
async def _load_asl_by_type(self, file: File, typ: ContentType) -> AsyncIterable[Tuple[str, Content]]:
session = get_session()
async with session.create_client('s3', region_name='us-east-1', aws_secret_access_key='moto',
aws_access_key_id='moto') as s3:
bucket_name = 'y-bucket'
await s3.create_bucket(Bucket=bucket_name)
await s3.put_object(Bucket=bucket_name, Key=file.path)
try:
async with AsyncFileIO(io=await s3.get_object(Bucket=bucket_name, Key=file.path)) as f:
file.in_progress()
collection = json.load(f)
for name, data in collection.items():
def _spec_load() -> Any:
_logger.info(f"Loading {typ.__qualname__} from {file.path}: {name}")
return spec.load(typ, data, key=name, options=self._options)
obj = await asyncio.get_running_loop().run_in_executor(None, _spec_load)
file.loaded()
yield name, obj
except Exception as error:
_logger.error(f"Failed to load file {file.path}: {error}")
file.failed()
raise error
</code>
@moto.mock_aws
async def _load_asl_by_type(self, file: File, typ: ContentType) -> AsyncIterable[Tuple[str, Content]]:
session = get_session()
async with session.create_client('s3', region_name='us-east-1', aws_secret_access_key='moto',
aws_access_key_id='moto') as s3:
bucket_name = 'y-bucket'
await s3.create_bucket(Bucket=bucket_name)
await s3.put_object(Bucket=bucket_name, Key=file.path)
try:
async with AsyncFileIO(io=await s3.get_object(Bucket=bucket_name, Key=file.path)) as f:
file.in_progress()
collection = json.load(f)
for name, data in collection.items():
def _spec_load() -> Any:
_logger.info(f"Loading {typ.__qualname__} from {file.path}: {name}")
return spec.load(typ, data, key=name, options=self._options)
obj = await asyncio.get_running_loop().run_in_executor(None, _spec_load)
file.loaded()
yield name, obj
except Exception as error:
_logger.error(f"Failed to load file {file.path}: {error}")
file.failed()
raise error
Traceback:
<code> raise ClientConnectorCertificateError(req.connection_key, exc) from exc
aiohttp.client_exceptions.ClientConnectorCertificateError: Cannot connect to host y-bucket.s3.amazonaws.com:443 ssl:True [SSLCertVerificationError: (1, '[SSL: CERTIFICATE_VERIFY_FAILED] certificate verify failed: unable to get local issuer certificate (_ssl.c:1133)')]
raise SSLError(endpoint_url=request.url, error=e)
botocore.exceptions.SSLError: SSL validation failed for https://y-bucket.s3.amazonaws.com/ Cannot connect to host y-bucket.s3.amazonaws.com:443 ssl:True [SSLCertVerificationError: (1, '[SSL: CERTIFICATE_VERIFY_FAILED] certificate verify failed: unable to get local issuer certificate (_ssl.c:1133)')]
</code>
<code> raise ClientConnectorCertificateError(req.connection_key, exc) from exc
aiohttp.client_exceptions.ClientConnectorCertificateError: Cannot connect to host y-bucket.s3.amazonaws.com:443 ssl:True [SSLCertVerificationError: (1, '[SSL: CERTIFICATE_VERIFY_FAILED] certificate verify failed: unable to get local issuer certificate (_ssl.c:1133)')]
raise SSLError(endpoint_url=request.url, error=e)
botocore.exceptions.SSLError: SSL validation failed for https://y-bucket.s3.amazonaws.com/ Cannot connect to host y-bucket.s3.amazonaws.com:443 ssl:True [SSLCertVerificationError: (1, '[SSL: CERTIFICATE_VERIFY_FAILED] certificate verify failed: unable to get local issuer certificate (_ssl.c:1133)')]
</code>
raise ClientConnectorCertificateError(req.connection_key, exc) from exc
aiohttp.client_exceptions.ClientConnectorCertificateError: Cannot connect to host y-bucket.s3.amazonaws.com:443 ssl:True [SSLCertVerificationError: (1, '[SSL: CERTIFICATE_VERIFY_FAILED] certificate verify failed: unable to get local issuer certificate (_ssl.c:1133)')]
raise SSLError(endpoint_url=request.url, error=e)
botocore.exceptions.SSLError: SSL validation failed for https://y-bucket.s3.amazonaws.com/ Cannot connect to host y-bucket.s3.amazonaws.com:443 ssl:True [SSLCertVerificationError: (1, '[SSL: CERTIFICATE_VERIFY_FAILED] certificate verify failed: unable to get local issuer certificate (_ssl.c:1133)')]
New contributor
kumaroid is a new contributor to this site. Take care in asking for clarification, commenting, and answering.
Check out our Code of Conduct.
I think it’d be a good idea to encapsulate the s3 interaction logic in an S3 client with a specific interface. I’ll use get_item_by_id instead of _load_asl_by_type method just to simplify the example:
<code>import uuid
from abc import ABC, abstractmethod
from pydantic import UUID4
class S3ClientInterface(ABC):
"""S3 client interface."""
@abstractmethod
async def get_item_by_id(self, item_id: UUID4) -> Item:
"""Get item by item id provided."""
class S3Client(S3ClientInterface):
"""S3 client."""
def __init__(self) -> None:
"""."""
self.session = ...
async def get_item_by_id(self, item_id: UUID4) -> Item:
"""Get item from real S3 by id provided."""
async with self.session.create_client(...):
...
return item
</code>
<code>import uuid
from abc import ABC, abstractmethod
from pydantic import UUID4
class S3ClientInterface(ABC):
"""S3 client interface."""
@abstractmethod
async def get_item_by_id(self, item_id: UUID4) -> Item:
"""Get item by item id provided."""
class S3Client(S3ClientInterface):
"""S3 client."""
def __init__(self) -> None:
"""."""
self.session = ...
async def get_item_by_id(self, item_id: UUID4) -> Item:
"""Get item from real S3 by id provided."""
async with self.session.create_client(...):
...
return item
</code>
import uuid
from abc import ABC, abstractmethod
from pydantic import UUID4
class S3ClientInterface(ABC):
"""S3 client interface."""
@abstractmethod
async def get_item_by_id(self, item_id: UUID4) -> Item:
"""Get item by item id provided."""
class S3Client(S3ClientInterface):
"""S3 client."""
def __init__(self) -> None:
"""."""
self.session = ...
async def get_item_by_id(self, item_id: UUID4) -> Item:
"""Get item from real S3 by id provided."""
async with self.session.create_client(...):
...
return item
Now you can make your “upper level” e.g. controller dependent from the S3ClientInterface, but not the S3Client:
<code>class ItemController:
"""Item controller."""
def __init__(self, s3_client: S3ClientInterface) -> None:
"""."""
self.s3_client = s3_client
async def process_item(self, item_id: UUID4) -> Item:
"""Get item from s3 and postprocess somehow."""
item = await self.s3_client.get_item_by_id(item_id=item_id)
# Do something with item
# Do more
return item
</code>
<code>class ItemController:
"""Item controller."""
def __init__(self, s3_client: S3ClientInterface) -> None:
"""."""
self.s3_client = s3_client
async def process_item(self, item_id: UUID4) -> Item:
"""Get item from s3 and postprocess somehow."""
item = await self.s3_client.get_item_by_id(item_id=item_id)
# Do something with item
# Do more
return item
</code>
class ItemController:
"""Item controller."""
def __init__(self, s3_client: S3ClientInterface) -> None:
"""."""
self.s3_client = s3_client
async def process_item(self, item_id: UUID4) -> Item:
"""Get item from s3 and postprocess somehow."""
item = await self.s3_client.get_item_by_id(item_id=item_id)
# Do something with item
# Do more
return item
And during testing your controller you can mock the whole S3 client with a testing one implementing the same interface:
<code>class MockS3Client(S3ClientInterface):
"""Mock S3 client."""
async def get_item_by_id(self, item_id: UUID4) -> Item:
"""Get mocked item from mocked S3 by id provided."""
# Return mocked item
return Item(
id=item_id,
name="Item",
size=5
)
# Tests
controller = ItemController(s3_client=MockS3Client())
controller.process_item(item_id=...) # Will use mocked get_item_by_id
</code>
<code>class MockS3Client(S3ClientInterface):
"""Mock S3 client."""
async def get_item_by_id(self, item_id: UUID4) -> Item:
"""Get mocked item from mocked S3 by id provided."""
# Return mocked item
return Item(
id=item_id,
name="Item",
size=5
)
# Tests
controller = ItemController(s3_client=MockS3Client())
controller.process_item(item_id=...) # Will use mocked get_item_by_id
</code>
class MockS3Client(S3ClientInterface):
"""Mock S3 client."""
async def get_item_by_id(self, item_id: UUID4) -> Item:
"""Get mocked item from mocked S3 by id provided."""
# Return mocked item
return Item(
id=item_id,
name="Item",
size=5
)
# Tests
controller = ItemController(s3_client=MockS3Client())
controller.process_item(item_id=...) # Will use mocked get_item_by_id