trovesuite

Shared service layer for TroveSuite applications. Bundles the internal building blocks — authentication, authorization, notifications, and Azure blob storage — behind a small, consistent API so downstream FastAPI services don't reimplement them.

This package is built for the TroveSuite platform and assumes TroveSuite's core_platform PostgreSQL schema (tenants, users, groups, roles, permissions). It is published to PyPI as a convenience for TroveSuite services; it is not a general-purpose auth library.

What's inside

Install

pip install trovesuite

Or with Poetry:

poetry add trovesuite

Requires Python 3.13.

Configure

The package reads configuration from environment variables (a .env file is loaded automatically via python-dotenv). See .env.example for the full list. The essentials:

# Postgres
DB_HOST=localhost
DB_PORT=5432
DB_NAME=trovesuite
DB_USER=postgres
DB_PASSWORD=...

# JWT
SECRET_KEY=change-me
ALGORITHM=HS256
ACCESS_TOKEN_EXPIRE_MINUTES=60

# Email (optional — used by Helper.send_notification)
MAIL_SENDER_EMAIL=alerts@yourdomain.com
MAIL_SENDER_PWD=gmail-app-password

# Azure Storage (optional — used by StorageService)
STORAGE_ACCOUNT_NAME=yourstorageaccount
USER_ASSIGNED_MANAGED_IDENTITY=<client-id>  # omit for system-assigned

Usage

Authorize a user from a JWT

from trovesuite import AuthService

result = AuthService.authorize_user_from_token(token)

if result.success:
    for role_entry in result.data:
        print(role_entry.role_id, role_entry.permissions)
else:
    print(result.error, result.detail)

Authorize by IDs

from trovesuite import AuthService
from trovesuite.auth import AuthServiceWriteDto

result = AuthService.authorize(
    AuthServiceWriteDto(user_id="usr_...", tenant_id="tnt_...")
)

authorize verifies the tenant exists and is verified, checks login-window restrictions (working days or custom time period), merges tenant-level and system-level role assignments, and returns a Respons[AuthServiceReadDto] with each role's permissions attached.

Possible result.error codes: INVALID_USER_ID, INVALID_TENANT_ID, TENANT_NOT_FOUND, TENANT_NOT_VERIFIED, USER_NOT_FOUND, USER_SUSPENDED, LOGIN_TIME_RESTRICTED, LOGIN_DAY_RESTRICTED.

Check permissions

from trovesuite import AuthService

# Does the user have 'permission-user-create'?
can_create = AuthService.check_permission(
    users_data=result.data,
    action="permission-user-create",
)

# Same check, but only against roles scoped to a specific resource type
can_create_in_group = AuthService.check_permission(
    users_data=result.data,
    action="permission-user-create",
    resource_type="rt-group",
)

# Aggregate helpers
all_perms = AuthService.get_user_permissions(result.data)
AuthService.has_any_permission(result.data, ["p-read", "p-write"])
AuthService.has_all_permissions(result.data, ["p-read", "p-write"])

Send an email

from trovesuite import NotificationService
from trovesuite.notification import NotificationEmailServiceWriteDto

result = NotificationService.send_email(NotificationEmailServiceWriteDto(
    sender_email="alerts@yourdomain.com",
    receiver_email=["user@example.com"],        # str or list[str]
    password="gmail-app-password",
    subject="Welcome",
    text_message="Plain text body",
    html_message="<h1>HTML body</h1>",          # optional
))

For tenant-aware sending that pulls credentials from the platform's cp_notification_email_credentials table (and falls back to MAIL_SENDER_*), use Helper.send_notification(...) with a template.

Azure Blob Storage

StorageService authenticates via Managed Identity (user-assigned if managed_identity_client_id is provided, otherwise falls back to DefaultAzureCredential for local dev). SAS URLs are generated via user-delegation keys — no storage account key required.

from trovesuite import StorageService
from trovesuite.storage import (
    StorageFileUploadServiceWriteDto,
    StorageFileUrlServiceWriteDto,
)

upload = StorageService.upload_file(StorageFileUploadServiceWriteDto(
    storage_account_url="https://myaccount.blob.core.windows.net",
    container_name="documents",
    blob_name="invoice.pdf",
    directory_path="tenants/tnt_abc",  # optional prefix
    file_content=pdf_bytes,
    content_type="application/pdf",
))

url = StorageService.get_file_url(StorageFileUrlServiceWriteDto(
    storage_account_url="https://myaccount.blob.core.windows.net",
    container_name="documents",
    blob_name="tenants/tnt_abc/invoice.pdf",
    expiry_hours=2,
))

Also available: create_container, update_file, delete_file, delete_multiple_files, download_file.

Response shape

Every service method returns a generic Respons[T]:

class Respons[T](BaseModel):
    detail: Optional[str]       # human-readable message
    error: Optional[str]        # machine error code (None on success)
    data: Optional[List[T]]     # payload; always a list, possibly empty
    status_code: int = 200
    success: bool = True
    pagination: Optional[PaginationMeta] = None

Always branch on result.success before reading result.data.

Database expectations

AuthService and Helper read from the TroveSuite core_platform schema. This package does NOT create or migrate the database — schema provisioning is owned by tvs-sqlscript (EF Core migrations). Run it once per environment before starting any consumer of this package:

cd ../tvs-sqlscript
dotnet run --project src/Trovesuite.Database.Runner -- deploy --module all

Table names are configurable via environment variables (CORE_PLATFORM_TENANTS_TABLE, CORE_PLATFORM_USER_GROUPS_TABLE, CORE_PLATFORM_LOGIN_SETTINGS_TABLE, CORE_PLATFORM_ASSIGN_ROLES_TABLE, CORE_PLATFORM_ROLE_PERMISSIONS_TABLE, CORE_PLATFORM_ROLES_TABLE, CORE_PLATFORM_USERS_TABLE, CORE_PLATFORM_ACTIVITY_LOGS_TABLE, CORE_PLATFORM_RESOURCE_ID_TABLE, CORE_PLATFORM_NOTIFICATION_EMAIL_CREDENTIALS_TABLE) — see .env.example.

Schemas must include the columns referenced in src/trovesuite/auth/auth_service.py (notably delete_status, is_active, is_system, is_suspended, working_days, login_on, logout_on, resource_type).

Development

git clone <repo>
cd tvs-package
poetry install --with dev

poetry run pytest
poetry run black src/
poetry run mypy src/
poetry run flake8 src/

Releasing

1. Bump version in pyproject.toml (both [tool.poetry] and [project] blocks) and setup.py.
2. Commit, then tag: git tag v$(new_version) && git push --tags.
3. The publish GitHub Actions workflow builds an sdist + wheel and uploads to PyPI using the PYPI_API_TOKEN secret from the pypi environment.

License

MIT — see LICENSE.