vanty-core 0.1.0

Shared base models, ULID fields, security middleware, and TaskIQ integration for the Vanty ecosystem

vanty-core

Shared base models, ULID fields, security middleware, and TaskIQ integration for the Vanty ecosystem.

Installation

pip install vanty-core

With optional dependencies:

# Auth integration (organization scoping, task context)
pip install vanty-core[auth]

# TaskIQ background tasks with auth context propagation
pip install vanty-core[taskiq]

# Everything
pip install vanty-core[all]

ULID Primary Keys

All base models use ULID primary keys – 128-bit identifiers that are lexicographically sortable by creation time. Stored as native UUID columns in the database.

from vanty_core.db.models import TimestampedModel
from tortoise import fields

class Product(TimestampedModel):
    name = fields.CharField(max_length=255)

    class Meta:
        table = "product"

Records are automatically assigned time-ordered IDs and can be sorted by id for chronological ordering.

Organization-Scoped Models

Models that belong to a tenant automatically filter queries based on the active AuthContext from vanty-auth:

from vanty_core.db.models import OrganizationScopedModel
from tortoise import fields

class Project(OrganizationScopedModel):
    name = fields.CharField(max_length=255)

    class Meta:
        table = "project"

# Automatically filtered to current organization
projects = await Project.filter(name__icontains="demo")

# Bypass scoping for admin operations
all_projects = await Project.unscoped.all()

If no AuthContext is active, scoped queries return empty results (fail-closed).

Security Middleware

Scanner Blocker

Blocks common vulnerability scanner paths at the ASGI level:

from vanty_core.security.middleware import SecurityScannerBlockerMiddleware

app.add_middleware(SecurityScannerBlockerMiddleware)

Rate Limiting

Factory for slowapi with in-memory, Redis, or Memcached backends:

from vanty_core.security.rate_limit import create_limiter
from slowapi import _rate_limit_exceeded_handler
from slowapi.errors import RateLimitExceeded

limiter = create_limiter(storage_uri="redis://localhost:6379")
app.state.limiter = limiter
app.add_exception_handler(RateLimitExceeded, _rate_limit_exceeded_handler)

@router.post("/submit")
@limiter.limit("10/minute")
async def submit(request: Request):
    ...

TaskIQ Integration

Propagate auth context into background tasks:

from vanty_core.tasks.broker import create_broker
from vanty_core.tasks.middleware import AuthContextTaskMiddleware
from vanty_core.tasks.context import send_with_context

broker = create_broker()
broker.add_middlewares([AuthContextTaskMiddleware(auth_context_service)])

@broker.task
async def process_order(order_id: str):
    # AuthContext is available here via asgiref.local
    ...

# Send with current auth context attached
await send_with_context(process_order, order_id="123")

Development

uv sync --dev
uv run pytest -q
uv run ruff check

Publishing

Tags trigger the release workflow:

git tag v0.1.0
git push origin v0.1.0

The GitHub Actions pipeline builds, creates a GitHub release, and publishes to PyPI via trusted publishing (OIDC).