sharepoint-transcribe/sp-transcribe-service/lambdas/shared.py

"""Shared utilities for all Lambda functions."""
import os
import json
import time
import boto3

PROJECT = os.environ.get('PROJECT', 'sp-transcribe')
S3_BUCKET = os.environ.get('S3_BUCKET', '')
USERS_TABLE = os.environ.get('USERS_TABLE', '')
JOBS_TABLE = os.environ.get('JOBS_TABLE', '')
SSM_PREFIX = os.environ.get('SSM_PREFIX', '/sp-transcribe')

_ssm_cache = {}

def get_ssm(name: str) -> str:
    if name in _ssm_cache:
        return _ssm_cache[name]
    ssm = boto3.client('ssm')
    resp = ssm.get_parameter(Name=f"{SSM_PREFIX}/{name}", WithDecryption=True)
    val = resp['Parameter']['Value']
    _ssm_cache[name] = val
    return val

def get_user(api_key: str) -> dict | None:
    ddb = boto3.resource('dynamodb')
    table = ddb.Table(USERS_TABLE)
    resp = table.get_item(Key={'api_key': api_key})
    return resp.get('Item')

def update_job(job_id: str, **kwargs):
    ddb = boto3.resource('dynamodb')
    table = ddb.Table(JOBS_TABLE)
    expr_parts = []
    values = {}
    names = {}
    for k, v in kwargs.items():
        safe = k.replace('.', '_')
        expr_parts.append(f"#{safe} = :{safe}")
        values[f":{safe}"] = v
        names[f"#{safe}"] = k
    table.update_item(
        Key={'job_id': job_id},
        UpdateExpression="SET " + ", ".join(expr_parts),
        ExpressionAttributeValues=values,
        ExpressionAttributeNames=names,
    )

def api_response(status: int, body: dict) -> dict:
    return {
        'statusCode': status,
        'headers': {'Content-Type': 'application/json'},
        'body': json.dumps(body),
    }