Guides

# Storage
export EVENTKIT_EVENT_STORE="gcs"          # gcs | firestore
export GCP_PROJECT_ID="my-project"
export GCP_GCS_BUCKET="my-events"
export GCP_BIGQUERY_DATASET="events"
 
# Queue
export EVENTKIT_QUEUE_MODE="async"         # async | pubsub
export EVENTKIT_ASYNC_WORKERS="4"
 
# Ring Buffer (Durability)
export EVENTKIT_RING_BUFFER_DB_PATH="./eventkit_ring_buffer.db"

# Development
docker compose up -d                       # Start emulators
uv run uvicorn eventkit.api.app:app --reload
 
# Testing
pytest                                     # All tests
pytest tests/unit/                         # Unit tests only
pytest --cov=src/eventkit                  # With coverage
 
# Code quality
mypy src/eventkit                          # Type check
ruff check src/                            # Lint
ruff format src/                           # Format

# Health checks
curl http://localhost:8000/health          # Liveness
curl http://localhost:8000/ready           # Readiness
 
# Collection endpoints
POST /collect/{stream}                     # Flexible ingestion
POST /v1/identify                          # Segment-compatible
POST /v1/track
POST /v1/page

from eventkit.adapters.validators import Validator
 
class CustomValidator(Validator):
    def validate(self, payload: dict) -> tuple[bool, str | None]:
        # Your validation logic
        if "required_field" not in payload:
            return False, "Missing required_field"
        return True, None
 
# Use in pipeline
from eventkit.adapters.validators import ValidationPipeline
pipeline = ValidationPipeline([CustomValidator(), ...])

# Path 1: Real-time processing (<1s latency)
realtime_coordinator = EventSubscriptionCoordinator(
    subscription="events-realtime",
    handler=process_immediately
)
 
# Path 2: Batched archival (5-10 min latency)
archive_coordinator = EventSubscriptionCoordinator(
    subscription="events-archive",
    handler=event_loader.add
)
 
manager = EventCoordinatorManager([
    realtime_coordinator,
    archive_coordinator
])
await manager.start_all()