Skip to content

DEVELOPER.md

Latest commit

 

History

History
268 lines (199 loc) · 9.45 KB

DEVELOPER.md

File metadata and controls

268 lines (199 loc) · 9.45 KB
Raw

EventGate for Developers

Get Started

Clone the repository and navigate to the project directory:

git clone https://github.com/AbsaOSS/EventGate.git
cd EventGate

Project Structure

EventGate ships two Lambda functions:

  • Event Gate Lambda (src/event_gate_lambda.py) — the main API surface. Serves the OpenAPI spec, token provider redirect, health check, topic schema catalogue, and event ingestion (POST /topics/{topicName}).
  • Event Stats Lambda (src/event_stats_lambda.py) — serves read-only queries via POST /stats/{topicName} with filtering, sorting, and cursor-based pagination backed by PostgreSQL.

Prerequisites

  • Python 3.13 (current required runtime)
  • Docker (for local integration tests using testcontainers)

Set Up Python Environment

python3 -m venv .venv
source .venv/bin/activate
pip3 install -r requirements-dev.txt

Run Pylint Tool Locally

This project uses the Pylint tool for static code analysis. Pylint analyzes your code without actually running it. It checks for errors, enforces coding standards, looks for code smells, etc.

Pylint displays a global evaluation score for the code, rated out of a maximum score of 10.0. We aim to keep our code quality above 9.5.

Run Pylint

Run Pylint on all files currently tracked by Git in the project.

pylint $(git ls-files '*.py')

To run Pylint on a specific file, follow the pattern pylint <path_to_file>/<name_of_file>.py.

Example:

pylint src/event_gate_lambda.py

Run Black Tool Locally

This project uses the Black tool for code formatting. Black aims for consistency, generality, readability, and reducing git diffs. The coding style used can be viewed as a strict subset of PEP 8.

The project root file pyproject.toml defines the Black tool configuration. In this project, we are accepting a line length of 120 characters.

Follow these steps to format your code with Black locally:

Run Black

Run Black on all files currently tracked by Git in the project.

black $(git ls-files '*.py')

To run Black on a specific file, follow the pattern black <path_to_file>/<name_of_file>.py.

Example:

black src/writers/writer_kafka.py

Expected Output

This is the console's expected output example after running the tool:

All done! ✨ 🍰 ✨
1 file reformatted.

Run mypy Tool Locally

This project uses the mypy tool, a static type checker for Python.

Type checkers help ensure that you correctly use variables and functions in your code. With mypy, add type hints (PEP 484) to your Python programs, and mypy will warn you when you use those types incorrectly. mypy configuration is in pyproject.toml file.

Follow these steps to type-check your code with mypy locally:

Run mypy

Run mypy on all files in the project.

mypy .

To run mypy on a specific file, follow the pattern mypy <path_to_file>/<name_of_file>.py --check-untyped-defs.

Example:

mypy src/handlers/handler_token.py

Run Unit Test Locally

Unit tests are written using pytest. To run the tests, use the following command:

pytest tests/unit/

This will execute all unit tests located in the tests/unit/ directory.

Focused / Selective Test Runs

Run a single test file:

pytest tests/unit/writers/test_writer_kafka.py

Filter by keyword expression:

pytest -k kafka

Run a single test function (node id):

pytest tests/unit/writers/test_writer_eventbridge.py::test_write_success

Code Coverage

Code coverage is collected using the pytest-cov coverage tool. To run the tests and collect coverage information, use the following command:

pytest --cov=. -v tests/unit/ --cov-fail-under=90 --cov-report=html

This will execute all tests in the tests directory and generate a code coverage report with missing line details and enforce a minimum 90% threshold.

Open the HTML coverage report:

open htmlcov/index.html

Run Integration Test Locally

Integration tests validate EventGate against real service dependencies using testcontainers-python.

Integration Test Approach

EventGate uses a direct invocation approach for integration testing:

  • Lambda handler is called directly in Python (not run in a container)
  • External dependencies run in Docker containers: Kafka, PostgreSQL, LocalStack (EventBridge)
  • Mock JWT provider runs in-process as a background thread (no container)
  • Test configuration is dynamically generated and injected via environment variables

Prerequisites

  • Docker running (Docker Desktop on macOS/Windows, or Docker Engine on Linux)
  • Python 3.13 with dependencies installed

When using colima on macOS, set the following environment variables before running tests:

export DOCKER_HOST=unix://$HOME/.colima/default/docker.sock
export TESTCONTAINERS_DOCKER_SOCKET_OVERRIDE=/var/run/docker.sock

Run Integration Tests

Containers start and stop automatically:

pytest tests/integration/ -v

With detailed logging:

pytest tests/integration/ -v --log-cli-level=INFO

Run Specific Integration Tests

Run a single test file:

pytest tests/integration/test_health_endpoint.py -v

Run a specific test function:

pytest tests/integration/test_topics_endpoint.py::TestPostEventEndpoint::test_post_event_with_valid_token_returns_202 -v

Troubleshooting

If containers fail to start, check Docker is running:

docker info

If image pulls fail with TLS or timeout errors, pre-pull the required images manually:

docker pull testcontainers/ryuk:0.8.1
docker pull postgres:16
docker pull confluentinc/cp-kafka:7.6.0
docker pull localstack/localstack:latest

View container logs in pytest output by increasing log level:

pytest tests/integration/ -v --log-cli-level=DEBUG

Logging Conventions

Logging is structured. src/utils/observability.py attaches the AWS Lambda Powertools JSON handler to the root logger, so modules keep using the standard library logger and inherit the format, the Lambda execution context and the request correlation id.

import logging

logger = logging.getLogger(__name__)

Rules:

  • The message is a constant sentence ending with a period. Variable data goes into extra, never into the sentence. 
    logger.warning("Request rejected: unknown topic.", extra={"known_topics": sorted(known_topics)})
  • Do not add topic, user, resource, http_method, correlation_id or the Lambda context to extra; they are bound once per request by bind_request_context() and append_request_context().
  • Every non-2xx response must produce exactly one log line explaining the cause.
  • Every request produces exactly one INFO line: Request completed., emitted by dispatch_request(). Handlers do not emit their own INFO outcome lines; they attach outcome fields with append_request_context() (e.g. writers_ok, message_key, row_count) so the completion line carries them.
  • Every failed request produces exactly one ERROR record (the aggregated dispatch failure); per-writer failure detail is logged at WARNING. This keeps level = "ERROR" metric filters counting failures, not log lines. The per-writer WARNING carries exc_info=True, because the aggregated ERROR is emitted outside the except block and can no longer reach the traceback.
  • Levels: TRACE payloads, DEBUG steps, INFO request outcomes, WARNING rejected requests and soft failures, ERROR failures that need action. See the level table in README.
  • Never log tokens, passwords or full message payloads outside TRACE. TRACE payload logging goes through log_payload_at_trace(), which redacts and size caps the payload.
  • logger.exception() is only valid inside an except block. Outside one, pass the captured exception: logger.error("...", exc_info=exc).
  • Durations are logged as milliseconds with an explicit key (duration_ms, writer_duration_ms, query_duration_ms).

Assert on structured fields in tests, not on formatted strings:

def test_rejects_unknown_topic(caplog):
    caplog.set_level(logging.WARNING)
    ...
    assert "Request rejected: unknown topic." == caplog.records[-1].message

Keys bound with append_request_context() live on the Powertools formatter rather than on the LogRecord. To assert on them, render the record with the Powertools logger (registered_formatter exists only there, not on a logging.getLogger() instance):

from src.utils.observability import logger as powertools_logger

payload = json.loads(powertools_logger.registered_formatter.format(caplog.records[-1]))
assert "run-42" == payload["correlation_id"]

Run All Quality Gates

Run Black, Pylint, mypy, unit tests (with coverage), and integration tests in a single command:

make qa

The command executes each gate in order and stops on first failure. Individual targets are also available (e.g., make black, make pylint, make pytest-unit).