Implement boto3 session use case and add tests

This commit is contained in:
Roman Mogylatov 2021-03-06 10:12:59 -05:00
parent 35a6f47aff
commit dbc6393561
3 changed files with 51 additions and 12 deletions

View File

@ -150,18 +150,21 @@ class SchemaProcessorV1:
return None return None
for segment in segments[1:]: for segment in segments[1:]:
if segment == 'as_int()': parentheses = ''
provider = provider.as_int() if '(' in segment and ')' in segment:
elif segment == 'as_float()': parentheses = segment[segment.find('('):segment.rfind(')')+1]
provider = provider.as_float() segment = segment.replace(parentheses, '')
elif segment.startswith('is_'): # TODO
provider = provider.as_(str) try:
... provider = getattr(provider, segment)
else: except AttributeError:
try: # TODO
provider = getattr(provider, segment) return None
except AttributeError:
return None if parentheses:
# TODO
provider = provider()
return provider return provider

View File

@ -0,0 +1,26 @@
version: "1"
container:
config:
provider: Configuration
session:
provider: Singleton
provides: boto3.session.Session
kwargs:
aws_access_key_id: container.config.aws_access_key_id
aws_secret_access_key: container.config.aws_secret_access_key
aws_session_token: container.config.aws_session_token
s3_client:
provider: Factory
provides: container.session.provided.client.call()
kwargs:
service_name: s3
sqs_client:
provider: Factory
provides: container.session.provided.client.call()
kwargs:
service_name: sqs

View File

@ -273,3 +273,13 @@ class TestSchemaMultipleContainersWithInlineProviders(unittest.TestCase):
self.assertIs(photo_service1.s3, photo_service2.s3) self.assertIs(photo_service1.s3, photo_service2.s3)
self.assertIs(photo_service1.s3, container.gateways.s3_client()) self.assertIs(photo_service1.s3, container.gateways.s3_client())
self.assertIs(photo_service2.s3, container.gateways.s3_client()) self.assertIs(photo_service2.s3, container.gateways.s3_client())
class TestSchemaBoto3Session(unittest.TestCase):
def test(self):
container = containers.DynamicContainer()
container.from_yaml_schema(f'{_SAMPLES_DIR}/schemasample/container-boto3-session.yml')
self.assertEqual(container.s3_client().__class__.__name__, 'S3')
self.assertEqual(container.sqs_client().__class__.__name__, 'SQS')