prisma-web3-py 1.0.10

Python/SQLAlchemy async implementation of Prisma Web3 database models

Prisma Web3 Python

Async PostgreSQL ORM layer for HawkFi and downstream bots, built on SQLAlchemy 2.x async. Models mirror the Prisma schema (managed in the JS project) and expose repositories plus 2.x-style session helpers.

Contents

• Tables & Repositories
• Session & Config
• Install
• Usage
• Testing
• CI / GitHub Actions
• Cleanup
• Best Practices

Tables & Repositories

Model (table)	Repository	Purpose / notable fields
Token ("Token")	TokenRepository	Chain-normalized token catalog; search/fuzzy/alias; batch upsert (chain, token_address, symbol, aliases, platforms).
Signal ("Signal")	SignalRepository	Token alerts; source, signal_type, occurrence_count, last_occurrence.
PreSignal ("PreSignal")	PreSignalRepository	Early signals; heat scores (channel_calls, multi_signals, kol_discussions).
CryptoNews ("CryptoNews")	CryptoNewsRepository	News ingestion with unique source_link; JSONB fields matched_currencies, entity_list.
AIAnalysisResult ("AIAnalysisResult")	AIAnalysisRepository	Unified analysis (Twitter/News/etc.); JSONB heavy (tokens, key_points, final_signal, trading_recommendation, event_struct); JSONB search/stats helpers.
EventImpacts ("EventImpacts")	EventImpactRepository	Post-event pricing snapshots (price_t0/30m/4h/24h, deltas, meta).
EventLabels ("EventLabels")	EventLabelsRepository	Label catalog for events.

Utilities: ChainConfig (chain normalization), datetime helpers, TokenImporter.

Session & Config

• URL: DATABASE_URL (or TEST_DATABASE_URL in tests). postgres:///postgresql:// auto-normalized to postgresql+asyncpg://.
• Pools: DB_POOL_SIZE, DB_POOL_OVERFLOW, DB_POOL_TIMEOUT.
• Engine/Session helpers (SQLAlchemy 2.x async):

from prisma_web3_py import session_scope, configure_engine, dispose_engine

# Optionally override URL (e.g., tests)
configure_engine("postgresql+asyncpg://user:pass@localhost:5432/prisma_web3")

async def main():
    async with session_scope() as session:  # commits on success, rollback on error
        ...
    async with session_scope(readonly=True) as session:  # read-only, no commit
        ...

Install

# editable dev install
pip install -e python[dev]

# or published package
pip install prisma-web3-py

Usage

from prisma_web3_py import session_scope, TokenRepository

repo = TokenRepository()

async def create_token():
    async with session_scope() as session:
        await repo.upsert_token(session, {"chain": "eth", "token_address": "0x...", "symbol": "UNI"})

async def search():
    async with session_scope(readonly=True) as session:
        tokens = await repo.search_by_symbol(session, "UNI")

Testing

• Unit-only (no DB required):

cd python
pytest -q tests/test_ai_analysis_utils.py tests/test_config_and_datetime.py tests/test_base_repository_unit.py

• Full + integration (requires Postgres):

export DATABASE_URL=postgresql+asyncpg://user:pass@localhost:5432/prisma_web3_test
export RUN_DB_TESTS=1
cd python
pytest        # integration tests will run

• Schema setup in CI: call python - <<'PY' to run init_db() (fallback create_all) if Prisma migrations aren’t applied yet.

CI / GitHub Actions

prisma-web3/.github/workflows/publish.yml runs unit + integration tests before publish. Key envs:

• DATABASE_URL – set via repo/org secrets for self-hosted DB or use Postgres service.
• RUN_DB_TESTS=1 – enables integration suite.

Postgres service example (already in workflow):

DATABASE_URL=postgresql+asyncpg://prisma:prisma@localhost:5432/prisma_web3_test

Integration step creates schema via init_db() prior to pytest.

Cleanup

python/scripts/cleanup_test_data.py deletes test tokens/signals/pre-signals using session_scope.

Best Practices

• Use session_scope/get_db (alias) for unit-of-work; avoid manual commits in callers.
• Keep repositories thin; heavy normalization belongs in services. AIAnalysis repo currently mixes normalization with persistence—plan to extract to service layer.
• JSONB/pg_trgm features are Postgres-specific; document for consumers.
• Prefer ORM select().scalars(), insert().on_conflict_do_update(); avoid relying on rowcount to detect changes.
• Production deployments should run Prisma migrations; init_db() is a fallback for tests/local only.