Polymorphic (de)serialization support (#7104, PR #7127)

Author: achave11-ucsc

.mypy.ini

@@ -55,7 +55,11 @@ modules =
     azul.service.app_controller,
     azul.indexer.mirror_service,
     azul.digests,
-    lambdas.indexer.app
+    lambdas.indexer.app,
+    azul.schemas,
+    azul.indexer.index_controller,
+    azul.indexer.index_queue_service,
+    azul.indexer.index_repository_service
 packages =
     azul.openapi
 

lambdas/indexer/app.py

@@ -1,4 +1,3 @@
-import json
 import logging
 from typing import (
     Optional,
@@ -7,14 +6,9 @@
 import chalice
 
 from azul import (
-    CatalogName,
     JSON,
     cached_property,
     config,
-    mutable_furl,
-)
-from azul.azulclient import (
-    IndexAction,
 )
 from azul.chalice import (
     LambdaMetric,
@@ -43,15 +37,6 @@
 )
 from azul.openapi import (
     format_description as fd,
-    params,
-    responses,
-    schema,
-)
-from azul.openapi.responses import (
-    json_content,
-)
-from azul.queues import (
-    Queues,
 )
 from azul.types import (
     not_none,
@@ -84,14 +69,7 @@ def index_controller(self) -> IndexController:
 
     @cached_property
     def mirror_controller(self) -> MirrorController:
-
-        def schema_url(*, schema_name: str, version: int) -> mutable_furl:
-            path = self.schema_url_path.format(facility='mirror',
-                                               schema_name=schema_name,
-                                               version_and_extension=f'v{version}.json')
-            return self.base_url.set(path=path)
-
-        return MirrorController(app=self, schema_url_func=schema_url)
+        return MirrorController(app=self)
 
     @cached_property
     def log_controller(self) -> LogForwardingController:
@@ -124,13 +102,6 @@ def decorator(f):
         else:
             return lambda func: func
 
-    schema_url_path = '/schemas/{facility}/{schema_name}/{version_and_extension}'
-
-    def schema_resource(self, *path: str) -> JSON:
-        schema = json.loads(app.load_static_resource('schemas', *path))
-        schema['$id'] = str(self.self_url)
-        return schema
-
     def _authenticate(self) -> Optional[HMACAuthentication]:
         return self.auth_from_request(not_none(self.current_request))
 
@@ -140,153 +111,7 @@ def _authenticate(self) -> Optional[HMACAuthentication]:
 
 globals().update(app.default_routes())
 
-
-@app.route(
-    '/{catalog}/{action}',
-    methods=['POST'],
-    spec={
-        'tags': ['Indexing'],
-        'summary': 'Notify the indexer to perform an action on a bundle',
-        'description': fd('''
-            Queue a bundle for addition to or deletion from the index.
-
-            The request must be authenticated using HMAC via the ``signature``
-            header. Each Azul deployment has its own unique HMAC key. The HMAC
-            components are the request method, request path, and the SHA256
-            digest of the request body.
-
-            A valid HMAC header proves that the client is in possession of the
-            secret HMAC key and that the request wasn't tampered with while
-            travelling between client and service, even though the latter is not
-            strictly necessary considering that TLS is used to encrypt the
-            entire exchange. Internal clients can obtain the secret key from the
-            environment they are running in, and that they share with the
-            service. External clients must have been given the secret key. The
-            now-defunct DSS was such an external client. The Azul indexer
-            provided the HMAC secret to DSS when it registered with DSS to be
-            notified about bundle additions/deletions. These days only internal
-            clients use this endpoint.
-        '''),
-        'requestBody': {
-            'description': 'Contents of the notification',
-            'required': True,
-            **json_content(schema.object(
-                bundle_fqid=schema.object(
-                    uuid=str,
-                    version=str,
-                    source=schema.object(
-                        id=str,
-                        spec=str
-                    )
-                )
-            ))
-        },
-        'parameters': [
-            params.path('catalog',
-                        schema.enum(*config.catalogs),
-                        description='The name of the catalog to notify.'),
-            params.path('action',
-                        schema.enum(IndexAction.add.name, IndexAction.delete.name),
-                        description='Which action to perform.'),
-            params.header('signature',
-                          str,
-                          description='HMAC authentication signature.')
-        ],
-        'responses': {
-            '200': {
-                'description': 'Notification was successfully queued for processing'
-            },
-            '400': {
-                'description': 'Request was rejected due to malformed parameters'
-            },
-            '401': {
-                'description': 'Request lacked a valid HMAC header'
-            }
-        }
-    }
-)
-def post_notification(catalog: CatalogName, action: str):
-    """
-    Receive a notification event and queue it for indexing or deletion.
-    """
-    return app.index_controller.handle_notification(catalog, action)
-
-
-@app.metric_alarm(metric=LambdaMetric.errors,
-                  threshold=int(config.contribution_concurrency(retry=False) * 2 / 3),
-                  period=5 * 60)
-@app.metric_alarm(metric=LambdaMetric.throttles,
-                  threshold=int(96000 / config.contribution_concurrency(retry=False)),
-                  period=5 * 60)
-@app.on_sqs_message(
-    queue=config.notifications_queue.name,
-    batch_size=1
-)
-def contribute(event: chalice.app.SQSEvent):
-    app.index_controller.contribute(event)
-
-
-@app.metric_alarm(metric=LambdaMetric.errors,
-                  threshold=int(config.aggregation_concurrency(retry=False) * 3),
-                  period=5 * 60)
-@app.metric_alarm(metric=LambdaMetric.throttles,
-                  threshold=int(37760 / config.aggregation_concurrency(retry=False)),
-                  period=5 * 60)
-@app.on_sqs_message(
-    queue=config.tallies_queue.name,
-    batch_size=Queues.batch_size
-)
-def aggregate(event: chalice.app.SQSEvent):
-    app.index_controller.aggregate(event)
-
-
-# Any messages in the tallies queue that fail being processed will be retried
-# with more RAM in the tallies_retry queue.
-
-@app.metric_alarm(metric=LambdaMetric.errors,
-                  threshold=int(config.aggregation_concurrency(retry=True) * 1 / 16),
-                  period=5 * 60)
-@app.metric_alarm(metric=LambdaMetric.throttles,
-                  threshold=0,
-                  period=5 * 60)
-@app.on_sqs_message(
-    queue=config.tallies_queue.to_retry.name,
-    batch_size=Queues.batch_size
-)
-def aggregate_retry(event: chalice.app.SQSEvent):
-    app.index_controller.aggregate(event, retry=True)
-
-
-# Any messages in the notifications queue that fail being processed will be
-# retried with more RAM and a longer timeout in the notifications_retry queue.
-
-@app.metric_alarm(metric=LambdaMetric.errors,
-                  threshold=int(config.contribution_concurrency(retry=True) * 1 / 4),
-                  period=5 * 60)
-@app.metric_alarm(metric=LambdaMetric.throttles,
-                  threshold=int(31760 / config.contribution_concurrency(retry=True)),
-                  period=5 * 60)
-@app.on_sqs_message(
-    queue=config.notifications_queue.to_retry.name,
-    batch_size=1
-)
-def contribute_retry(event: chalice.app.SQSEvent):
-    app.index_controller.contribute(event, retry=True)
-
-
-if config.enable_mirroring:
-    @app.metric_alarm(metric=LambdaMetric.errors,
-                      threshold=int(config.mirroring_concurrency * 2 / 3),
-                      period=5 * 60)
-    @app.metric_alarm(metric=LambdaMetric.throttles,
-                      threshold=int(96000 / config.mirroring_concurrency),
-                      period=5 * 60)
-    @app.on_sqs_message(
-        queue=config.mirror_queue.name,
-        batch_size=1
-    )
-    def mirror(event: chalice.app.SQSEvent):
-        app.mirror_controller.mirror(event)
+globals().update(app.index_controller.handlers())
 
 
 @app.log_forwarder(
@@ -303,40 +128,4 @@ def forward_s3_logs(event: chalice.app.S3Event):
     app.log_controller.forward_s3_access_logs(event)
 
 
-json_schema_docs = 'https://json-schema.org/docs'
-
-
-@app.route(
-    app.schema_url_path,
-    methods=['GET'],
-    cors=True,
-    spec={
-        'summary': 'Retrieve JSON schemas',
-        'tags': ['Auxiliary'],
-        'parameters': [
-            params.path('facility', str),
-            params.path('schema_name', str),
-            params.path('version_and_extension', schema.pattern(r'v\d+\.json')),
-        ],
-        'description': fd(
-            f'''
-            [JSON Schemas]({json_schema_docs}) for various Azul facilities.
-            '''
-        ),
-        'responses': {
-            '200': {
-                'description': 'Contents of the schema',
-                **responses.json_content(
-                    schema.object(
-                        schema=str,
-                        id=str,
-                        type=str,
-                        additionalProperties=True
-                    )
-                )
-            }
-        }
-    }
-)
-def get_schema(facility, schema_name, version_and_extension):
-    return app.schema_resource(facility, schema_name, version_and_extension)
+globals().update(app.mirror_controller.handlers())

scripts/mirror_file.py

@@ -66,24 +66,23 @@ def mirror_file(catalog: CatalogName, file_uuid: str, part_size: int) -> str:
     file = get_file(catalog, file_uuid)
     # FIXME: mirror_file script is broken+
     #        https://github.com/DataBiosphere/azul/issues/7105
-    service = MirrorService(schema_url_func=...)
-    upload_id = service.begin_mirroring_file(catalog, file)
+    service = MirrorService(catalog=catalog, schema_url_func=...)
+    upload_id = service.begin_mirroring_file(file)
     digest_value, digest_type = file.digest()
     hasher = get_resumable_hasher(digest_type)
 
     def mirror_parts():
         part = FilePart.first(file, part_size)
         while part is not None:
-            yield service.mirror_file_part(catalog, file, part, upload_id, hasher)
+            yield service.mirror_file_part(file, part, upload_id, hasher)
             part = part.next(file)
 
     etags = list(mirror_parts())
-    service.finish_mirroring_file(catalog=catalog,
-                                  file=file,
+    service.finish_mirroring_file(file=file,
                                   upload_id=upload_id,
                                   etags=etags,
                                   hasher=hasher)
-    return service.get_mirror_url(catalog, file)
+    return service.get_mirror_url(file)
 
 
 def main(argv):

scripts/reindex.py

@@ -175,7 +175,7 @@ def main(argv: list[str]):
                 if args.local:
                     num_notifications += azul.local_reindex(catalog, args.prefix)
                 else:
-                    azul.remote_reindex(catalog, sources)
+                    azul.index_queue_service.remote_reindex(catalog, sources)
                     num_notifications = None
             else:
                 log.info('Skipping catalog %r (no matching sources)', catalog)