Skip to main content

Backend Development

This guide covers development patterns for the VAMS Python Lambda backend, including handler structure, Pydantic model definitions, two-tier authorization, Amazon DynamoDB access patterns, and error handling.

Technology Stack​

ComponentDetails
RuntimePython 3.12 (AWS Lambda)
ValidationPydantic v1 only, via aws-lambda-powertools
AuthorizationCasbin ABAC/RBAC with Amazon DynamoDB policy storage
AWS SDKboto3
SearchOpenSearch (opensearch-py)
LoggingAWS Lambda Powertools Logger with custom redaction
Testingpytest, moto

Exact versions are pinned in backend/requirements.txt and backend/requirements-dev.txt, which are generated from poetry.lock. Install with pip install -r requirements.txt rather than pinning packages by hand.

Pydantic v1 Only

VAMS uses Pydantic v1 (1.10.x). Never use Pydantic v2 syntax (model_validator, model_dump, ConfigDict). Import BaseModel from aws_lambda_powertools.utilities.parser, not from pydantic directly. Violations cause import failures in Lambda.

Project Structure​

backend/
  backend/
    common/                          # Shared utilities
      constants.py                   # ABAC policy, allowed values, file blocklists
      dynamodb.py                    # DynamoDB helpers (to_update_expr, get_asset_object_from_id)
      validators.py                  # Input validation regex patterns and validate() dispatcher
      s3.py                          # S3 file validation (extension + MIME type checks)
      s3MetadataKeys.py              # Canonical S3 object user-metadata keys (assetid, vams-*)
      s3PathPatterns.py              # Reserved S3 prefixes, preview file pattern, preview extensions
      dynamoDbMetadataKeys.py        # Special DynamoDB metadata keys and internal field prefixes
    customLogging/
      auditLogging.py                # CloudWatch audit logging (9 event types)
      logger.py                      # safeLogger wrapper with sensitive data redaction
    handlers/                        # Lambda handlers (one folder per domain)
      assets/assetService.py         # Gold standard handler
      auth/                          # Auth handlers (authorizer, constraints, cognito)
      authz/__init__.py              # Casbin ABAC/RBAC enforcer (CasbinEnforcer)
      databases/                     # Database CRUD
      metadata/                      # Metadata CRUD
      pipelines/                     # Pipeline management
      workflows/                     # Step Functions workflow management
      ...                            # Additional handler domains
    models/                          # Pydantic v1 models
      assetsV3.py                    # Gold standard model file
      common.py                      # Response helpers, error functions
      pipelines.py                   # Pipeline models
      workflows.py                   # Workflow models
      ...                            # Domain-specific models
  tests/                             # Test suite
    mocks/                           # Mock modules replacing real imports

Gold Standard Handler Pattern​

Every new Lambda handler must follow the structure demonstrated in backend/backend/handlers/assets/assetService.py. The pattern consists of five layers.

1. Module-Level Setup​

Set up imports, AWS clients, logger, and environment variables at the module level. This code executes once during Lambda cold start.

import os
import boto3
import json
from botocore.config import Config
from aws_lambda_powertools.utilities.typing import LambdaContext
from aws_lambda_powertools.utilities.parser import parse, ValidationError
from common.constants import STANDARD_JSON_RESPONSE
from common.validators import validate
from handlers.authz import CasbinEnforcer
from handlers.auth import request_to_claims
from customLogging.logger import safeLogger
from models.common import (
    APIGatewayProxyResponseV2, internal_error, success,
    validation_error, general_error, authorization_error,
    VAMSGeneralErrorResponse
)
from models.yourDomain import YourRequestModel
from common.resourceNames import ResourceKeys, get_table_name

# Configure AWS clients with retry configuration
retry_config = Config(retries={'max_attempts': 5, 'mode': 'adaptive'})
dynamodb = boto3.resource('dynamodb', config=retry_config)
dynamodb_client = boto3.client('dynamodb', config=retry_config)
logger = safeLogger(service_name="YourServiceName")

claims_and_roles = {}

try:
    your_table_name = get_table_name(ResourceKeys.YOUR_STORAGE_TABLE)
except Exception as e:
    logger.exception("Failed loading environment variables")
    raise e

your_table = dynamodb.Table(your_table_name)
Resource Name Resolution

DynamoDB table, S3 bucket, and audit log group names are resolved at module level inside a try/except block through common.resourceNames (get_table_name, get_bucket_name, get_log_group_name with a ResourceKeys constant). The resolver checks a legacy environment-variable override first, then a cached batched AWS Systems Manager Parameter Store lookup under the deployment's VAMS_RESOURCE_PARAM_PREFIX. Non-resource configuration (function names, queue URLs, feature flags) still comes from os.environ at module level. Never resolve names inside handler functions.

2. Lambda Handler Entry Point​

The entry point extracts claims, performs API-level authorization, and routes to method handlers.

def lambda_handler(event, context: LambdaContext) -> APIGatewayProxyResponseV2:
    global claims_and_roles
    claims_and_roles = request_to_claims(event)

    try:
        method = event['requestContext']['http']['method']

        method_allowed_on_api = False
        if len(claims_and_roles["tokens"]) > 0:
            casbin_enforcer = CasbinEnforcer(claims_and_roles)
            if casbin_enforcer.enforceAPI(event):
                method_allowed_on_api = True

        if not method_allowed_on_api:
            return authorization_error()

        if method == 'GET':
            return handle_get(event)
        elif method == 'PUT':
            return handle_put(event)
        elif method == 'DELETE':
            return handle_delete(event)
        else:
            return validation_error(body={'message': "Method not allowed"}, event=event)

    except ValidationError as v:
        logger.exception(f"Validation error: {v}")
        return validation_error(body={'message': str(v)}, event=event)
    except VAMSGeneralErrorResponse as v:
        logger.exception(f"VAMS error: {v}")
        return general_error(body={'message': str(v)}, event=event)
    except Exception as e:
        logger.exception(f"Internal error: {e}")
        return internal_error(event=event)

3. Method Handlers​

Route HTTP methods to specific business logic functions based on the request path.

def handle_get(event):
    path = event['requestContext']['http']['path']
    query_params = event.get('queryStringParameters', {}) or {}

    if '/items/' in path:
        item_id = path.split('/items/')[-1]
        return get_single_item(event, item_id)
    else:
        return get_all_items(event, query_params)

4. Business Logic Functions​

Each business logic function follows a four-step pattern: validate, query, authorize, respond.

def get_single_item(event, item_id):
    # Step 1: Validate input parameters
    (valid, message) = validate({
        'itemId': {'value': item_id, 'validator': 'ID'}
    })
    if not valid:
        return validation_error(body={'message': message}, event=event)

    # Step 2: Query DynamoDB
    response = your_table.get_item(Key={'itemId': item_id})
    item = response.get('Item')
    if not item:
        return general_error(body={'message': 'Item not found'}, event=event)

    # Step 3: Object-level authorization
    item['object__type'] = 'yourObjectType'
    casbin_enforcer = CasbinEnforcer(claims_and_roles)
    if not casbin_enforcer.enforce(item, "GET"):
        return authorization_error()

    # Step 4: Return response
    return success(body=item)

5. Error Handling Hierarchy​

ExceptionResponse FunctionHTTP Status
ValidationError (Pydantic)validation_error()400
VAMSGeneralErrorResponsegeneral_error()400
Authorization failureauthorization_error()403
Exception (catch-all)internal_error()500

All response functions accept an optional event= parameter for audit logging. Always pass the event when available.

Two-Tier Authorization​

VAMS enforces authorization at two levels. Both levels must allow access for a request to succeed.

Tier 1: API-Level Authorization​

Controls which API routes a role can access. Performed in the lambda_handler using enforceAPI().

casbin_enforcer = CasbinEnforcer(claims_and_roles)
if not casbin_enforcer.enforceAPI(event):
    return authorization_error()

Tier 2: Object-Level Authorization​

Controls which specific data entities a role can access. Performed in business logic functions using enforce(obj, act), where obj is the entity dictionary and act is the HTTP method the caller must be allowed to perform on it (GET, POST, PUT, DELETE).

# MUST annotate the object type before calling enforce()
item['object__type'] = 'asset'
casbin_enforcer = CasbinEnforcer(claims_and_roles)
if not casbin_enforcer.enforce(item, "GET"):
    return authorization_error()

Only enforceAPI() takes the Lambda event; enforce() never does. Passing the event as the first argument evaluates an object with no constraint fields and denies every request.

Object Type Annotation

You must add object__type to the item dictionary before calling enforce(). Valid object types include: database, asset, api, web, tag, tagType, role, userRole, pipeline, workflow, metadataSchema, apiKey.

Key Authorization Concepts​

  • CasbinEnforcer uses a 60-second policy cache TTL per user
  • Policy is stored in Amazon DynamoDB (ConstraintsStorageTable)
  • Claims are extracted via request_to_claims(event) which returns user tokens, roles, and MFA status
  • Roles with mfaRequired=True are only active when mfaEnabled=True in claims

Pydantic v1 Model Patterns​

Reference file: backend/backend/models/assetsV3.py

Correct Model Definition​

from typing import Dict, List, Optional
from pydantic import Field
from aws_lambda_powertools.utilities.parser import (
    BaseModel, root_validator, validator, ValidationError
)
from common.validators import validate, id_pattern, object_name_pattern

class CreateItemRequestModel(BaseModel, extra='ignore'):
    """Request model for creating a new item"""
    databaseId: str = Field(
        min_length=4, max_length=256,
        strip_whitespace=True, regex=id_pattern
    )
    itemName: str = Field(
        min_length=1, max_length=256,
        strip_whitespace=True, regex=object_name_pattern
    )
    description: str = Field(min_length=4, max_length=256, strip_whitespace=True)
    tags: Optional[list[str]] = []

    @root_validator
    def validate_fields(cls, values):
        (valid, message) = validate({
            'tags': {
                'value': values.get('tags'),
                'validator': 'STRING_256_ARRAY',
                'optional': True
            }
        })
        if not valid:
            raise ValueError(message)
        return values

Common Mistakes to Avoid​

# WRONG: Importing from pydantic directly
from pydantic import BaseModel

# WRONG: Using Pydantic v2 syntax
class MyModel(BaseModel):
    model_config = ConfigDict(extra='ignore')    # v2 syntax

# WRONG: Missing extra='ignore'
class MyModel(BaseModel):
    pass

# WRONG: Using model_validate or model_dump (v2)
item = MyModel.model_validate(data)
data = item.model_dump()

# CORRECT alternatives:
from aws_lambda_powertools.utilities.parser import parse
item = parse(body, model=MyModel)
data = item.dict()

Parsing Request Bodies​

from aws_lambda_powertools.utilities.parser import parse

body = json.loads(event.get('body', '{}'))
request = parse(body, model=CreateItemRequestModel)

Amazon DynamoDB Patterns​

Table Initialization​

# Module-level: resource API for high-level operations
from common.resourceNames import ResourceKeys, get_table_name

dynamodb = boto3.resource('dynamodb', config=retry_config)
your_table = dynamodb.Table(get_table_name(ResourceKeys.YOUR_STORAGE_TABLE))

# Module-level: client API for low-level operations
dynamodb_client = boto3.client('dynamodb', config=retry_config)

Common Operations​

# Query with key condition
from boto3.dynamodb.conditions import Key

response = your_table.query(
    KeyConditionExpression=(
        Key('databaseId').eq(database_id) & Key('assetId').eq(asset_id)
    ),
    ScanIndexForward=False
)

# Get single item
response = your_table.get_item(Key={'itemId': item_id})
item = response.get('Item')

# Put item with condition
your_table.put_item(
    Item=item_dict,
    ConditionExpression='attribute_not_exists(databaseId) and attribute_not_exists(itemId)'
)

# Update item using the to_update_expr helper
from common.dynamodb import to_update_expr

keys_map, values_map, expr = to_update_expr(update_dict)
your_table.update_item(
    Key={'itemId': item_id},
    UpdateExpression=expr,
    ExpressionAttributeNames=keys_map,
    ExpressionAttributeValues=values_map
)

Pagination Pattern​

VAMS uses Base64-encoded NextToken pagination:

import base64

def get_paginated_items(event, query_params):
    max_items = int(query_params.get('maxItems', '100'))
    next_token = query_params.get('NextToken')

    scan_kwargs = {'Limit': max_items}

    if next_token:
        decoded = json.loads(base64.b64decode(next_token).decode('utf-8'))
        scan_kwargs['ExclusiveStartKey'] = decoded

    response = your_table.scan(**scan_kwargs)
    items = response.get('Items', [])

    result = {'Items': items}
    if 'LastEvaluatedKey' in response:
        result['NextToken'] = base64.b64encode(
            json.dumps(response['LastEvaluatedKey']).encode('utf-8')
        ).decode('utf-8')

    return success(body=result)

Input Validation​

Use the validate() dispatcher from common.validators for all input validation, both in @root_validator methods and in handler code.

from common.validators import validate

(valid, message) = validate({
    'databaseId': {'value': database_id, 'validator': 'ID'},
    'assetId': {'value': asset_id, 'validator': 'ASSET_ID'},
    'tags': {
        'value': tag_list,
        'validator': 'STRING_256_ARRAY',
        'optional': True
    }
})
if not valid:
    return validation_error(body={'message': message}, event=event)

Available Validators​

ValidatorPatternUse For
ID^[-_a-zA-Z0-9]{3,63}$databaseId, pipelineId
ASSET_IDfilename pattern, max 256assetId
UUIDStandard UUID formatUnique identifiers
OBJECT_NAME^[a-zA-Z0-9\-._\s]{1,256}$assetName, dbName
EMAILEmail regexEmail addresses
USERID^[\w\-\.\+\@]{3,256}$User identifiers
FILE_NAMENo special charactersFile names
STRING_256Max 256 charsMedium strings
STRING_16384Max 16384 charsFree-form caller text (comment bodies)
ID_ARRAYArray of IDsMultiple IDs
STRING_256_ARRAYArray of max-256 stringsTags, lists
ARNPartition-aware AWS ARNAny AWS resource ARN (sub-process registration)
CLOUDWATCH_LOG_GROUP_ARNPartition-aware log-group ARNRegistered CloudWatch log-group locations
CLOUDWATCH_LOG_GROUP_NAME1-512 chars (-_./# + alnum)Registered CloudWatch log-group names
LOG_STREAM_NAME1-512 chars, no : or *Registered log-stream names / prefixes

All AWS-resource validators are partition-aware (commercial, GovCloud, China, ISO).

The dispatcher recognizes only the names it implements, and the _VALIDATOR_DISPATCH mapping in common/validators.py is that list — a new validation type is one entry in it, with no second list to update. A name with no entry has no rule to apply, so it is rejected rather than reported valid unchecked. The name is resolved after the empty/optional short-circuits, so an optional field left empty is skipped before its validator is consulted.

Regex Patterns for Pydantic Fields​

from common.validators import (
    id_pattern,                  # r'^[-_a-zA-Z0-9]{3,63}$'
    filename_pattern,            # For asset IDs and file names
    object_name_pattern,         # r'^[a-zA-Z0-9\-._\s]{1,256}$'
    relative_file_path_pattern,  # r'^\/.*$'
)

Shared Constants​

System-owned key names, path prefixes, and file-name patterns are defined once in dedicated modules under backend/backend/common/. Always import these constants instead of redefining the literal values at call sites, so all usages can be found and changed in one place.

ModuleDefines
s3MetadataKeys.pyS3 object user-metadata keys (assetid, databaseid, vams-* status and change-provenance keys)
s3PathPatterns.pyReserved S3 prefix folders, the .previewFile. marker, allowed preview extensions, write prefixes
dynamoDbMetadataKeys.pySpecial DynamoDB metadata keys (REINDEX_METADATA_RECORD) and internal field prefixes (VAMS_, _)

S3 Path Patterns​

common/s3PathPatterns.py is the single source of truth for the reserved S3 folder names and the file-level preview file pattern. The indexers, bucket sync, workflow auto-trigger, and add-on syncs (Garnet, Physna) all consume these values when deciding which S3 keys to skip.

from common.s3PathPatterns import (
    RESERVED_S3_PREFIX_FOLDERS,       # frozenset: pipeline(s), preview(s), temp-upload(s), workspace(s)
    EXCLUDED_FILE_PATH_PATTERNS,      # patterns excluded from generic file processing
    PREVIEW_FILE_PATTERN,             # '.previewFile.' marker substring
    ALLOWED_PREVIEW_FILE_EXTENSIONS,  # ('.png', '.jpg', '.jpeg', '.svg', '.gif')
    TEMPORARY_UPLOAD_PREFIX,          # 'temp-uploads/'
    PREVIEW_PREFIX,                   # 'previews/' (asset bucket)
    PIPELINES_PREFIX,                 # 'pipelines/' (workflow run I/O in the default bucket)
    AUXILIARY_PREVIEW_PREFIX,         # 'preview/' (auxiliary bucket, singular)
)

Pipeline staging paths follow the structure pipelines/{pipelineName}/{jobName}/output/{executionId}/{outputType}/, relative to the area VAMS owns in the default asset bucket. executionRecords.run_bucket_key() joins that bucket's baseAssetsPrefix onto a relative key to produce the key an Amazon S3 call uses, and returns the key unchanged for a bucket registered at the root. The workflow state machine carries the relative form and the bucket's prefix as separate values, so a definition never embeds the prefix. The path segments within this structure are also defined as constants:

from common.s3PathPatterns import (
    PIPELINE_OUTPUT_PREFIX,           # '/output/' (required by the ASSET_PATH_PIPELINE validator)
    PIPELINE_INPUT_PREFIX,            # '/input/' (reserved for a future feature)
    PIPELINE_OUTPUT_FILES_PREFIX,     # '/files/' (file-level outputs, outputS3AssetFilesPath)
    PIPELINE_OUTPUT_PREVIEWS_PREFIX,  # '/previews/' (asset-level previews, outputS3AssetPreviewPath)
    PIPELINE_OUTPUT_METADATA_PREFIX,  # '/metadata/' (metadata files, outputS3AssetMetadataPath)
    PIPELINE_OUTPUT_RESULTS_PREFIX,   # '/results/' (structured pipeline result files, recorded by the end-state lambda)
)
Frontend Mirror

The preview file pattern and allowed preview extensions are mirrored in the frontend at web/src/common/constants/fileFormats.ts (PREVIEW_FILE_PATTERN, previewFileFormats). Keep the two in sync when changing them.

DynamoDB Metadata Keys​

common/dynamoDbMetadataKeys.py defines the special key names VAMS reserves inside its metadata storage tables. The reindexer writes a REINDEX_METADATA_RECORD marker item to trigger stream processing, and every consumer that reads metadata items must skip the system record keys.

from common.dynamoDbMetadataKeys import (
    REINDEX_METADATA_RECORD_KEY,    # reindexer touch marker (writer side)
    EXCLUDED_METADATA_RECORD_KEYS,  # frozenset of all system record keys to skip
    VAMS_INTERNAL_FIELD_PREFIX,     # 'VAMS_' prefix on internal asset-metadata fields
    HIDDEN_FIELD_PREFIX,            # '_' prefix; excluded from search and export output
    is_excluded_metadata_record,    # helper: is this key a system record to skip?
    is_internal_metadata_field,     # helper: is this field VAMS-internal?
)

When reading metadata items, check is_excluded_metadata_record(key) rather than comparing against individual key constants. Future system keys added to EXCLUDED_METADATA_RECORD_KEYS are then picked up by every call site automatically. Writers of a specific marker record (such as the reindexer) still reference the individual key constant.

Logging​

safeLogger​

Use safeLogger from customLogging.logger for all logging. Never use print() or logging.getLogger().

from customLogging.logger import safeLogger

logger = safeLogger(service_name="YourServiceName")

logger.info("Processing request")
logger.error(f"Failed to process: {error_message}")
logger.exception(f"Unexpected error: {e}")   # Includes stack trace
logger.warning(f"Potential issue: {details}")

The logger automatically redacts sensitive fields at all nesting levels, in both objects and arrays:

  • authorization
  • idJwtToken
  • Credentials, AccessKeyId, SecretAccessKey, SessionToken
  • configBody, templateTags, tagValues -- caller-authored template content, also filtered inside a JSON-string request body

Redaction is driven by the field name, so a message that interpolates a payload value into a formatted string is written as-is. Log identifiers, counts, and flags rather than rendered bodies or tag values.

Audit Logging​

Nine dedicated Amazon CloudWatch log groups capture security-sensitive operations. See the Audit Logging guide for details.

Response Functions​

All handlers must use the standardized response functions from models/common.py:

from models.common import (
    success,              # 200
    validation_error,     # 400 -- validation failures
    general_error,        # 400 -- business logic errors
    authorization_error,  # 403 -- access denied
    internal_error,       # 500 -- unexpected errors
    VAMSGeneralErrorResponse  # Exception class for business logic
)

# Raise in business logic:
raise VAMSGeneralErrorResponse("Error getting bucket details.")

# Return from handlers:
return success(body={'items': items})
return validation_error(body={'message': 'Invalid ID format'}, event=event)

Adding a New API Endpoint​

Adding a new endpoint requires coordinated changes across multiple files.

Checklist​

StepFileAction
1backend/backend/handlers/{domain}/{handler}.pyImplement Lambda handler
2backend/backend/models/{domain}.pyDefine Pydantic v1 models
3infra/lib/lambdaBuilder/{domain}Functions.tsBuild Lambda with env vars and permissions
4infra/lib/nestedStacks/apiLambda/apiBuilder-nestedStack.tsAttach Lambda to API Gateway route
5web/src/services/APIService.tsAdd API call function
All Steps Required

A handler without an API Gateway route is dead code. A route without a handler returns HTTP 500. Always complete all steps when adding a new endpoint.

Handler Template​

import os
import boto3
import json
from botocore.config import Config
from aws_lambda_powertools.utilities.typing import LambdaContext
from aws_lambda_powertools.utilities.parser import parse, ValidationError
from common.validators import validate
from handlers.authz import CasbinEnforcer
from handlers.auth import request_to_claims
from customLogging.logger import safeLogger
from models.common import (
    APIGatewayProxyResponseV2, internal_error, success,
    validation_error, general_error, authorization_error,
    VAMSGeneralErrorResponse
)
from common.resourceNames import ResourceKeys, get_table_name

retry_config = Config(retries={'max_attempts': 5, 'mode': 'adaptive'})
dynamodb = boto3.resource('dynamodb', config=retry_config)
logger = safeLogger(service_name="CHANGE_ME")

claims_and_roles = {}

try:
    table_name = get_table_name(ResourceKeys.CHANGE_ME_STORAGE_TABLE)
except Exception as e:
    logger.exception("Failed loading environment variables")
    raise e

table = dynamodb.Table(table_name)


def lambda_handler(event, context: LambdaContext) -> APIGatewayProxyResponseV2:
    global claims_and_roles
    claims_and_roles = request_to_claims(event)

    try:
        method = event['requestContext']['http']['method']

        method_allowed_on_api = False
        if len(claims_and_roles["tokens"]) > 0:
            casbin_enforcer = CasbinEnforcer(claims_and_roles)
            if casbin_enforcer.enforceAPI(event):
                method_allowed_on_api = True

        if not method_allowed_on_api:
            return authorization_error()

        if method == 'GET':
            return handle_get(event)
        elif method == 'PUT':
            return handle_put(event)
        elif method == 'DELETE':
            return handle_delete(event)
        else:
            return validation_error(body={'message': "Method not allowed"}, event=event)

    except ValidationError as v:
        logger.exception(f"Validation error: {v}")
        return validation_error(body={'message': str(v)}, event=event)
    except VAMSGeneralErrorResponse as v:
        logger.exception(f"VAMS error: {v}")
        return general_error(body={'message': str(v)}, event=event)
    except Exception as e:
        logger.exception(f"Internal error: {e}")
        return internal_error(event=event)

Custom Authentication Hooks​

VAMS provides two customization points for organizations to extend authentication behavior without modifying core code. Both files are located in backend/backend/customConfigCommon/.

Login Profile Customization​

The file customAuthLoginProfile.py controls how user profile information is updated when a user authenticates. Override the customAuthProfileLoginWriteOverride() function to customize profile data.

Default behavior: Extracts the email claim from the JWT token and writes it to the user's VAMS profile. The login profile is updated via an authenticated POST call to /api/auth/loginProfile/{userId} from the web UI on each login.

Common customizations:

  • Fetching additional user attributes from an external identity provider API
  • Populating the name field from directory services
  • Enriching the profile with organizational metadata
# backend/backend/customConfigCommon/customAuthLoginProfile.py
def customAuthProfileLoginWriteOverride(userProfile, lambdaRequestEvent):
    # Default: override email from JWT claims
    claims = ...  # extracted from request context
    if 'email' in claims:
        userProfile["email"] = claims['email']

    # Add custom logic here (e.g., fetch from external IDP userinfo endpoint)
    return userProfile
Email Fallback

The email field is used by systems that send notifications to the user. If the email is blank or not in a valid email format, VAMS falls back to using the userId as the notification address.

MFA and Claims Check Customization​

The file customAuthClaimsCheck.py controls how authentication claims are verified, including Multi-Factor Authentication (MFA) status.

The MFA check runs once at authorization time: the API Gateway custom authorizer calls customMFATokenScopeCheckOverride after verifying the caller's JWT and passes the result to handler Lambda functions as the vams:mfaEnabled authorizer context value. Handler Lambda functions read that context value in request_to_claims — they make no identity provider calls of their own.

Default behavior for Amazon Cognito: Resolves the user's MFA preference with the Cognito AdminGetUser API, cached per user per sign-in session (auth_time).

Default behavior for external OAuth IDP: Sets mfaEnabled to false. Organizations implement their own MFA verification logic (for example, a call to the IDP userinfo endpoint using the bearer token from the authorizer event headers) in the marked section of the hook.

# backend/backend/customConfigCommon/customAuthClaimsCheck.py
def customMFATokenScopeCheckOverride(user, authorizerJwtClaims, lambdaRequest):
    # Called by the API Gateway authorizer after JWT verification
    # For Cognito: checks UserMFASettingList via the AdminGetUser API
    # For external IDP: returns False by default
    # Override with your organization's MFA verification logic
    return mfaLoginEnabled

def customAuthClaimsCheckOverride(claims_and_roles, lambdaRequest):
    # Called by handler lambdas; mfaEnabled is already resolved from the
    # vams:mfaEnabled authorizer context value before this hook runs
    # Add additional handler-time claims validation logic here
    return claims_and_roles
Performance Consideration

customMFATokenScopeCheckOverride runs inside the API Gateway authorizer on every non-cached authorization. Cache external lookups (the default implementation caches by auth_time) and minimize external API calls to avoid adding latency to every request.

Notification Subscriptions​

handlers/subscription/subscriptionService.py records who is notified when an asset changes. Each entry is keyed on the event name and the entity (Asset#{assetId}) and holds a subscribers list; the handler resolves each subscriber to an email address and creates an Amazon SNS email subscription on the asset's own topic (AssetTopic{databaseId}-{assetId}). Notification content is the asset name and its current version identifier, sent by handlers/sendEmail/sendEmail.py.

How a subscriber value becomes an email address:

  1. subscribers is validated with the USERID_ARRAY validator, so each entry matches ^[\w\-\.\+\@]{3,256}$. The list has no maximum length.
  2. get_userProfile_Email() reads the entry as a userId in the user table. When a record exists with a non-empty email, that address is used.
  3. When no record exists, or the record's email is blank, the submitted value itself is checked with the EMAIL validator and used verbatim when it is email-shaped. A value that is neither a user with an email nor email-shaped is rejected with 400.
A subscriber does not have to be a VAMS user

Step 3 is deliberate — it allows a shared mailbox or resource account that has no VAMS identity to receive asset notifications — and it means the recipient list is not bounded by the user directory. Authorization covers the asset, not the recipients: the caller must pass the API-route tier for /subscriptions and the object tier for a POST on the target asset, and nothing further constrains whose address is added.

Two properties follow, and both are worth knowing before extending this handler:

  • Amazon SNS sends a confirmation request to each new address and delivers nothing until it is confirmed, so an unrecognized address receives one confirmation email and no asset data.
  • Subscriber values are stored as submitted (user identifiers or addresses), while the topic holds the resolved addresses — and the removal paths do not agree on which form they match. PUT /subscriptions unsubscribes by resolved address, and DELETE /subscriptions deletes the asset's topic outright, but DELETE /unsubscribe matches the submitted value against the topic's endpoints. A user whose profile email differs from their user identifier is therefore dropped from the subscription record while their topic subscription remains.

Subscription management is an administrative form. The Subscription Management page (/auth/subscriptions) is the only place an arbitrary recipient list is authored, and it is reachable only by a role granted that web route — the seeded admin role, whose default constraint allows all web paths. No shipped role template in documentation/permissionsTemplates/ grants it. The asset details pane offers an ordinary user a self-subscribe toggle, which submits only that user's own identifier; the /subscriptions API routes themselves carry no such restriction, so a role granted POST /subscriptions (the database-user template grants it) can submit any recipient through the API or the CLI.

Anti-Patterns​

Avoid these common mistakes in backend development.

Anti-PatternCorrect Approach
from pydantic import BaseModelfrom aws_lambda_powertools.utilities.parser import BaseModel
Raw dict responses {'statusCode': 200, ...}Use success(), validation_error(), etc.
print() for loggingUse logger.info(), logger.error()
Creating boto3 clients inside functionsCreate at module level with retry_config
Skipping enforceAPI() in handlerAlways check both auth tiers
Missing object__type before enforce()Annotate item before object-level auth
Inline regex validationUse validate() dispatcher
Swallowing exceptions with bare except: passLog errors and raise VAMSGeneralErrorResponse

Next Steps​