feat(stream_processor): initial version for processing product changes

.gitignore

@@ -252,3 +252,4 @@ lambda_requirements.txt
 # Misc
 
 node_modules
+.idea

Makefile

@@ -14,6 +14,10 @@ format:
 	poetry run isort .
 	poetry run yapf -d -vv --style=./.style -r .
 
+format-fix:
+	poetry run isort .
+	poetry run yapf -vv --style=./.style -r --in-place .
+
 lint: format
 	@echo "Running flake8"
 	poetry run flake8 product/* infrastructure/* tests/* docs/examples/*
@@ -68,3 +72,6 @@ docs:
 
 lint-docs:
 	docker run -v ${PWD}:/markdown 06kellyjac/markdownlint-cli --fix "docs"
+
+watch:
+	npx cdk watch

cdk.json

@@ -1,3 +1,16 @@
 {
-  "app": "poetry run python app.py"
+  "app": "poetry run python app.py",
+  "watch": {
+    "include": [
+      "product/**",
+      "infrastructure/product/**"
+    ],
+    "exclude": [
+      "product/**/*.pyc",
+      "product/**/__pycache__",
+      "infrastructure/**/*.pyc",
+      "infrastructure/**/__pycache__"
+    ]
+  },
+  "build": "make build"
 }

docs/api/product_models.md

@@ -0,0 +1,7 @@
+## Product models
+
+::: product.models.products.product
+
+## Validators
+
+::: product.models.products.validators

docs/api/stream_processor.md

@@ -0,0 +1,16 @@
+
+## Product notification
+
+### Lambda Handlers
+
+Process stream is connected to Amazon DynamoDB Stream that polls product changes in the product table.
+
+We convert them into `ProductChangeNotification` model depending on the DynamoDB Stream Event Name (e.g., `INSERT` -> `ADDED`).
+
+::: product.stream_processor.handlers.process_stream
+
+### Domain logic
+
+Domain logic to notify product changes, e.g., `ADDED`, `REMOVED`, `UPDATED`.
+
+::: product.stream_processor.domain_logic.product_notification

infrastructure/product/constants.py

@@ -1,5 +1,4 @@
 SERVICE_ROLE_ARN = 'ServiceRoleArn'
-STREAM_PROC_SERVICE_ROLE_ARN = 'StreamRoleArn'
 LAMBDA_BASIC_EXECUTION_ROLE = 'AWSLambdaBasicExecutionRole'
 CREATE_PRODUCT_ROLE = 'ServiceRole'
 DELETE_PRODUCT_ROLE = 'DeleteRole'
@@ -28,8 +27,11 @@
 BUILD_FOLDER = '.build/lambdas/'
 COMMON_LAYER_BUILD_FOLDER = '.build/common_layer'
 CRUD_CONSTRUCT_NAME = 'Crud'
-STREAM_PROC_CONSTRUCT_NAME = 'StreamProc'
 OWNER_TAG = 'owner'
 REST_API_NAME = 'crud-rest-api'
 EVENT_BUS_NAME = 'events'
-STREAM_PROC_LAMBDA = 'Stream'
+STREAM_PROCESSOR_CONSTRUCT_NAME = 'StreamProc'
+STREAM_PROCESSOR_LAMBDA = 'StreamProcessor'
+STREAM_PROCESSOR_LAMBDA_MEMORY_SIZE = 128  # MB
+STREAM_PROCESSOR_LAMBDA_TIMEOUT = 120  # seconds
+STREAM_PROCESSOR_LAMBDA_SERVICE_ROLE_ARN = 'StreamRoleArn'

infrastructure/product/service_stack.py

@@ -25,7 +25,7 @@ def __init__(self, scope: Construct, id: str, **kwargs) -> None:
 
         self.stream_processor = StreamProcessorConstruct(
             self,
-            id_=get_construct_name(id, constants.STREAM_PROC_CONSTRUCT_NAME),
+            id_=constants.STREAM_PROCESSOR_CONSTRUCT_NAME,
             lambda_layer=self.shared_layer,
             dynamodb_table=self.api.api_db.db,
         )

infrastructure/product/stream_processor_construct.py

@@ -17,14 +17,14 @@ def __init__(self, scope: Construct, id_: str, lambda_layer: PythonLayerVersion,
         super().__init__(scope, id_)
         bus_name = f'{id_}{constants.EVENT_BUS_NAME}'
         self.event_bus = events.EventBus(self, bus_name, event_bus_name=bus_name)
-        self.role = self._build_lambda_role(dynamodb_table, self.event_bus)
+        self.role = self._build_lambda_role(db=dynamodb_table, bus=self.event_bus)
 
         self.lambda_function = self._build_stream_processor_lambda(self.role, lambda_layer, dynamodb_table)
 
     def _build_lambda_role(self, db: dynamodb.Table, bus: events.EventBus) -> iam.Role:
         return iam.Role(
             self,
-            constants.STREAM_PROC_SERVICE_ROLE_ARN,
+            id=constants.STREAM_PROCESSOR_LAMBDA_SERVICE_ROLE_ARN,
             assumed_by=iam.ServicePrincipal('lambda.amazonaws.com'),
             inline_policies={
                 'streams':
@@ -51,21 +51,21 @@ def _build_lambda_role(self, db: dynamodb.Table, bus: events.EventBus) -> iam.Ro
     def _build_stream_processor_lambda(self, role: iam.Role, lambda_layer: PythonLayerVersion, dynamodb_table: dynamodb.Table) -> _lambda.Function:
         lambda_function = _lambda.Function(
             self,
-            constants.STREAM_PROC_LAMBDA,
+            id=constants.STREAM_PROCESSOR_LAMBDA,
             runtime=_lambda.Runtime.PYTHON_3_11,
             code=_lambda.Code.from_asset(constants.BUILD_FOLDER),
-            handler='product.stream_processor.handlers.stream_handler.handle_events',
+            handler='product.stream_processor.handlers.process_stream.process_stream',
             environment={
                 constants.POWERTOOLS_SERVICE_NAME: constants.SERVICE_NAME,  # for logger, tracer and metrics
                 constants.POWER_TOOLS_LOG_LEVEL: 'DEBUG',  # for logger
             },
             tracing=_lambda.Tracing.ACTIVE,
             retry_attempts=0,
-            timeout=Duration.seconds(constants.API_HANDLER_LAMBDA_TIMEOUT),
-            memory_size=constants.API_HANDLER_LAMBDA_MEMORY_SIZE,
+            timeout=Duration.seconds(constants.STREAM_PROCESSOR_LAMBDA_TIMEOUT),
+            memory_size=constants.STREAM_PROCESSOR_LAMBDA_MEMORY_SIZE,
             layers=[lambda_layer],
             role=role,
-            log_retention=RetentionDays.ONE_DAY,
+            log_retention=RetentionDays.FIVE_DAYS,
         )
         # Add DynamoDB Stream as an event source for the Lambda function
         lambda_function.add_event_source(DynamoEventSource(dynamodb_table, starting_position=_lambda.StartingPosition.LATEST))

infrastructure/product/utils.py

@@ -1,4 +1,4 @@
-import os
+import getpass
 from pathlib import Path
 
 from git import Repo
@@ -8,7 +8,7 @@
 
 def get_username() -> str:
     try:
-        return os.getlogin().replace('.', '-')
+        return getpass.getuser().replace('.', '-')
     except Exception:
         return 'github'
 
@@ -17,7 +17,7 @@ def get_stack_name() -> str:
     repo = Repo(Path.cwd())
     username = get_username()
     try:
-        branch_name = f'{repo.active_branch}'.replace('/', '-')
+        branch_name = f'{repo.active_branch}'.replace('/', '-').replace('_', '-')
         return f'{username}-{branch_name}-{constants.SERVICE_NAME}'
     except TypeError:
         # we're running in detached mode (HEAD)

mkdocs.yml

@@ -0,0 +1,99 @@
+site_name: Serverless Python Demo OPN305
+site_description: Serverless Python Demo for re:Invent OPN305 session
+site_author: Ran Isenberg and Heitor Lessa
+repo_url: https://github.com/ran-isenberg/serverless-python-demo
+edit_uri: edit/main/docs
+
+nav:
+  - Homepage:
+      - Decision log: decision_log.md
+      - API reference:
+          - Stream Processor: api/stream_processor.md
+          - Product models: api/product_models.md
+
+theme:
+  name: material
+  font:
+    text: Ubuntu
+  palette:
+    - scheme: default
+      primary: deep purple
+      toggle:
+        icon: material/lightbulb
+        name: Switch to dark mode
+    - scheme: slate
+      primary: indigo
+      accent: teal
+      toggle:
+        icon: material/lightbulb-outline
+        name: Switch to light mode
+  features:
+    - header.autohide
+    - navigation.sections
+    - navigation.top
+    - navigation.instant
+    - navigation.indexes
+    - navigation.tracking
+    - navigation.tabs
+    - content.code.annotate
+    - content.code.copy
+  icon:
+    repo: fontawesome/brands/github
+
+markdown_extensions:
+  - admonition
+  - abbr
+  - pymdownx.tabbed:
+      alternate_style: true
+  - pymdownx.highlight:
+      linenums: true
+  - pymdownx.details
+  - pymdownx.snippets:
+      base_path: "."
+      check_paths: true
+      restrict_base_path: false
+  - meta
+  - toc:
+      permalink: true
+      toc_depth: 4
+  - attr_list
+  - pymdownx.emoji:
+      emoji_index: !!python/name:materialx.emoji.twemoji
+      emoji_generator: !!python/name:materialx.emoji.to_svg
+  - pymdownx.inlinehilite
+  - pymdownx.superfences:
+      custom_fences:
+        - name: mermaid
+          class: mermaid
+          format: !!python/name:pymdownx.superfences.fence_code_format
+  - pymdownx.tasklist:
+      custom_checkbox: true
+
+copyright: Copyright © 2023 Ran Isenberg and Heitor Lessa
+
+plugins:
+  - search
+  - mkdocstrings:
+      handlers:
+        python:
+          options:
+            docstring_style: numpy
+            docstring_section_style: spacy
+            show_source: true
+            heading_level: 4
+            allow_inspection: true
+            group_by_category: true
+            show_category_heading: true
+            show_bases: true
+            show_docstring_examples: true
+            show_if_no_docstring: true
+            merge_init_into_class: true # avoid Class params + __init__ params
+            separate_signature: false
+            show_signature_annotations: false
+
+extra_css:
+  - stylesheets/extra.css
+
+watch:
+  - product
+  - docs

mypy.ini

@@ -7,7 +7,12 @@ warn_unused_ignores=True
 show_column_numbers = True
 show_error_codes = True
 show_error_context = True
+plugins = pydantic.mypy
 
+[pydantic-mypy]
+init_forbid_extra = true
+init_typed = true
+warn_required_dynamic_aliases = true
 
 # Disable specific error codes in the 'tests' package
 [mypy-tests.*]
@@ -67,3 +72,6 @@ ignore_missing_imports = True
 
 [mypy-setuptools]
 ignore_missing_imports = True
+
+[mypy-pytest_socket]
+ignore_missing_imports = True