Metadata-Version: 2.4
Name: cz-powertools
Version: 0.0.7
Summary: CloudZero wrappers around the routing functionality in AWS powertools (https://github.com/aws-powertools/powertools-lambda-python)
Requires-Python: >=3.12
Description-Content-Type: text/markdown
Requires-Dist: aws-lambda-powertools
Requires-Dist: cz-common-python>=8.0.0
Requires-Dist: pydantic>=2.0.0
Requires-Dist: requests
Provides-Extra: dev
Requires-Dist: flake8-copyright; extra == "dev"
Requires-Dist: isort; extra == "dev"
Requires-Dist: mypy; extra == "dev"
Requires-Dist: pytest; extra == "dev"
Requires-Dist: pytest-cov; extra == "dev"
Requires-Dist: pytest-csv; extra == "dev"
Requires-Dist: pytest-env; extra == "dev"
Requires-Dist: pytest-mock; extra == "dev"
Requires-Dist: pytest-xdist; extra == "dev"
Requires-Dist: ruff>=0.8.4; extra == "dev"
Requires-Dist: wheel; extra == "dev"

[![Build and Test All Branches](https://github.com/Cloudzero/cz-powertools-lambda-python/actions/workflows/ci.yml/badge.svg)](https://github.com/Cloudzero/cz-powertools-lambda-python/actions/workflows/ci.yml)


## Project Overview

This project is a CloudZero specific wrapper around AWS Powertools for Lambda. It provides routing and event handlers that provide all the important middleware for CZ handlers. Through using these event handlers and routers, you can have a single lambda handler for multiple API routes. This can help with performance and costs by leveraging provisioned concurrency across multiple API routes.

Additionally, the AWS powertools functionality provides additional functionality:
- Validation of all header, path parameters, query parameters, and input body are done using Pydantic.
- OpenAPI and Swagger support


### Web API

For supporting our public APIs for our frontend, use the following classes:

| Class | Location | Description |
| - | - | - |
| `WebApiGatewayRestResolver` | `czpowertools.event_handler.web_api_event_handler` | This is the main event handler that resolves the event to the different route handlers. |
| `WebApiRouter` | `czpowertools.event_handler.router.web_api_router` | This is a sub-router that adds the `/organizations/<organizationId>` portion of the route and can add additional prefixes for routing for a specific resource type (i.e. for the *recommendations* resource `/organizations/<organizationId>/optimize/recommendations`). |

Here is an [example](examples/web_api/web_api.py) using sub-routers with the `WebApiGatewayRestResolver`. You can see them execute using [these unit tests](tests/unit/web_api_example/test_web_api.py).

All routes added using `WebApiGatewayRestResolver` will get the following functionality:
- Logging context with organization ID/user ID tags
- AWS Tracing
- Organization Path Authorization (ensuring the organization the user is logged in with matches the organization ID in the route path parameter `organizationId`)
- Request Validation (headers, path parameters, query parameters, body)
- Exception/Response handling
- Test Key handling

All routes added using `WebApiRouter` will have the `/organizations/<organizationId>` prefix added to all routes. **NOTE**: If you add routes directly with `WebApiGatewayRestResolver` then you must prefix the path with `/organizations/<organizationId>`.

**NOTE**: All routes added using `WebApiRouter` or `WebApiGatewayRestResolver` must expect to have a path parameter `organizationId`.

### Public/Programmatic API

For supporting our public APIs for our customers, use the following classes:

| Class | Location | Description |
| - | - | - |
| `PublicApiGatewayRestResolver` | `czpowertools.event_handler.public_api_event_handler` | This is the main event handler that resolves the event to the different route handlers. |
| `PublicApiRouter` | `czpowertools.event_handler.router.public_api_router` | This is a sub-router that allows adding prefixes for routing for a specific resource type (i.e. for the *recommendations* resource `/optimize/recommendations`). |

Here is an [example](examples/public_api/public_api.py) using sub-routers with the `PublicApiGatewayRestResolver`. You can see them execute using [these unit tests](tests/unit/public_api_example/test_public_api.py).

All routes added using `PublicApiGatewayRestResolver` will get the following functionality:
- Logging context with organization ID/user ID tags
- AWS Tracing
- Organization Authorization (ensuring the authorization contains the `cz_organization_id` field and adds it to the app context.
- Request Validation (headers, path parameters, query parameters, body)
- Exception/Response handling
- Test Key handling
- Idempotency Support
- Cache Control support.

**NOTE**: Unlike the decorators from `cz-common-python`, the `cz_organization_id` is not added as a parameter to the handler calls as this interferes with parameter validation (it assumes it is a query parameter) as well as OpenApi/Swagger support. Both `PublicApiRouter` and `PublicApiGatewayRestResolver` have a `cz_organization_id` property which will contain the organization ID for the current request.

## Auditing

Auditing is the middleware replacement for the handler-level decorators in `czc.audit` (`public_api_audit_event_publisher`, `programmatic_api_audit_event_publisher`, etc.). When enabled, an audit event is published to the CloudZero primary event bus (`cz-{namespace}-events-primary`) after every successful (2xx) mutating request (POST, PUT, PATCH, DELETE by default). Requests are never failed by auditing: publishing errors are logged and the response is returned unchanged.

**NOTE**: Auditing is currently **opt-in** for backwards compatibility. The next major version will enable it by default.

### Enabling auditing

The simplest way is through the API configuration, which works with both the builders and the resolvers directly:

```python
from czpowertools.api_builder import WebApiBuilder, WebApiConfiguration

app = WebApiBuilder(configuration=WebApiConfiguration(enable_auditing=True)).add_router(my_router).build()
```

The same flag exists on `PublicApiConfiguration`. The audit event type defaults to `configuration` and can be changed with the `audit_event_type` configuration field. The resolver picks the correct middleware for its API type:

| Resolver | Middleware | Actor (`user_id`) | Organization |
| - | - | - | - |
| `WebApiGatewayRestResolver` | `WebApiAuditMiddleware` | `user_id` from the authorizer | `organization_id` from the authorizer |
| `PublicApiGatewayRestResolver` | `PublicApiAuditMiddleware` | `api_key_name` from the authorizer (default `organization-api-key`) | `cz_organization_id` from the authorizer |

Alternatively, the middleware can be added manually — for example to audit only specific methods or to use a different event type:

```python
from czpowertools.middlewares.audit_middleware import WebApiAuditMiddleware

configuration = WebApiConfiguration(
    additional_middlewares=[WebApiAuditMiddleware(event_type='configuration', audited_http_methods=['POST', 'DELETE'])],
)
```

### Audit event contents

The audit `target` defaults to the last static segment of the matched route (`/organizations/<organizationId>/budgets/<budgetId>` produces target `budgets`) and the internal change data defaults to the request path and parsed request body. Route handlers can customize the event through the app/router context:

```python
@router.post('/')
def create_budget(organizationId: str, body: BudgetModel):
    router.append_context(
        audit_target='budget',  # override the target name
        audit_data={'budget_name': body.name},  # override the internal change data
    )
    return {'id': create(body)}


@router.post('/dry-run')
def dry_run_budget(organizationId: str, body: BudgetModel):
    router.append_context(publish_audit_data=False)  # suppress the audit event
    return {'valid': True}
```

Supported context keys: `audit_target`, `audit_data`, `audit_external_change_data`, and `publish_audit_data`. Setting `audit_data` to an empty dict redacts the change data: the event is still published, with `internal_change_data` set to `{'redacted': true}`, e.g. to keep a sensitive request body out of the audit trail. `audit_external_change_data` must contain **both** `url` (a fully-qualified URL) and `id` — events failing this validation are not published, and the middleware raises a Sentry alarm (`invalid-audited-lambda-response`, the same alarm name used by the `czc.audit` decorators).

Publishing failures never fail the request, but every dropped audit event raises a Sentry alarm: schema rejections and missing authorizer context use `invalid-audited-lambda-response`, and other publishing failures (e.g. EventBridge throttling or a missing `events:PutEvents` permission) use `failed-audit-event-publish`.

### Opting in or out on specific routes

Auditing does not have to be all-or-nothing. There are two ways to scope it to a subset of routes, depending on which default you want:

**Opt out per route** (recommended): enable auditing service-wide with `enable_auditing=True` and suppress it in the routes that should not be audited by appending `publish_audit_data=False` to the context, as in the `/dry-run` example above. This keeps the safer default — new mutating routes are audited automatically — and makes each exception explicit in the handler:

```python
@router.post('/validate')
def validate_budget(organizationId: str, body: BudgetModel):
    router.append_context(publish_audit_data=False)  # not a real mutation; skip the audit event
    return {'valid': True}
```

**Opt in per route**: leave `enable_auditing` off (the default) and attach the audit middleware to individual routes with the `middlewares` argument that Powertools supports on every route decorator:

```python
from czpowertools.middlewares.audit_middleware import WebApiAuditMiddleware

audit = WebApiAuditMiddleware()  # share one instance across routes


@router.post('/', middlewares=[audit])  # audited
def create_budget(organizationId: str, body: BudgetModel):
    return {'id': create(body)}


@router.post('/dry-run')  # not audited
def dry_run_budget(organizationId: str, body: BudgetModel):
    return {'valid': True}
```

Use `PublicApiAuditMiddleware` for `PublicApiGatewayRestResolver`-based APIs. Prefer the opt-out approach for anything security-relevant: with per-route opt-in, a forgotten `middlewares=[audit]` on a new route is a silent gap in the audit trail, whereas a forgotten opt-out only produces an extra event. Non-mutating methods (GET, etc.) are never audited either way, so read-only routes need no configuration at all.

### Migrating from the `czc.audit` decorators

The middleware reads the context keys above instead of the `audit_data` key the decorators expected in the handler **response** — and unlike the decorators, it does not strip `audit_data` from the response. When migrating a handler, remove `audit_data` from the returned body (otherwise it is serialized into the HTTP response sent to the client) and use `router.append_context(...)` instead. Also note the default target is derived from the route path; if downstream consumers match on the decorator's explicit target (or the derived `event_name`), set `audit_target` to the value the decorator used.

### Required Lambda execution role changes

Audit events are published with `events:PutEvents` to the `cz-{namespace}-events-primary` event bus, so the Lambda execution role of any function with auditing enabled must be granted that permission. With SAM, add the policy to the function (or to the shared role used by your API handler functions):

```yaml
  WebApiHandlerFunction:
    Type: AWS::Serverless::Function
    Properties:
      # ...
      Policies:
        - EventBridgePutEventsPolicy:
            EventBusName: !Sub cz-${Namespace}-events-primary
```

Or as a raw IAM policy statement on the execution role:

```yaml
        - Effect: Allow
          Action: events:PutEvents
          Resource: !Sub arn:aws:events:${AWS::Region}:${AWS::AccountId}:event-bus/cz-${Namespace}-events-primary
```

The function must also have the `NAMESPACE` environment variable set (used to resolve the event bus name); this is already standard for CloudZero features. Without the permission, requests still succeed — the failed publish is logged by the middleware — but the audit trail will be silently incomplete, so treat the role update as part of enabling auditing.
