Step 2 - Database Engine Setup
Configure the asynchronous SQLAlchemy engine and understand how ZCore handles connection pooling and query logging.
ZCore does not hide SQLAlchemy; it orchestrates it. In this step, we will initialize the DatabaseManager and understand how ZCore handles asynchronous connection pooling and structured query logging.
Configure the Database URL
Open your .env file. By default, ZCore uses an asynchronous SQLite database for local development.
# .env
DATABASE_URL=sqlite+aiosqlite:///task_manager.dbFor production, you would switch this to PostgreSQL (e.g., postgresql+asyncpg://user:pass@host:port/db). ZCore's DatabaseManager automatically adjusts connection pooling based on the database dialect.
Initialize the Engine in main.py
Open main.py. The CLI generated the initialization code for you, but let's understand what it does.
# main.py
from zcore import Kernel, db_manager, settings
# Initialize Database Manager
db_manager.init_app(config=settings.DATABASE)config: Accepts aDatabaseSettingsinstance (likesettings.DATABASE) or a dictionary.pool_size: The number of connections to keep open in the pool (Default: 5).max_overflow: How many connections to allow beyondpool_sizeunder heavy load (Default: 10).pool_recycle: Recycles connections after a specified timeout (Default: 1800s).pool_pre_ping: Tests connections before handing them out to prevent stale connection errors (Default: True).
The Query Timing Logger
Under the hood: ZCore attaches event listeners to the SQLAlchemy engine. Every time a query is executed, ZCore calculates the duration in milliseconds and logs it using structlog. It also intelligently filters out noisy system queries (like pg_catalog lookups) so your logs remain clean.
In rc.2, SQL query logging defaults to False to prevent console noise during development. You can enable it via .env or capture slow queries exclusively:
# .env
LOG_SQL_QUERIES=True # Enable logging all SQL queries
# OR keep it False and set a slow-query threshold:
# SLOW_QUERY_THRESHOLD_MS=100.0 # Only log queries exceeding 100msWhen query logging is enabled, your logs will look like this:
[info] sql_query sql=SELECT tasks.id FROM tasks params=() duration_ms=1.23Dependency Injection for Sessions
ZCore provides a ready-to-use FastAPI dependency for database sessions. You don't need to write a get_db generator function manually.
from zcore import SessionDep
from sqlalchemy.ext.asyncio import AsyncSession
@app.get("/health")
async def health_check(db: SessionDep):
# db is an AsyncSession injected by ZCore's DI container
return {"status": "healthy"}SessionDep is an Annotated[AsyncSession, Depends(get_db)]. It resolves the active request-scoped session from the IoCContainer.
Your database engine is configured and optimized. In Step 3, we will scaffold our first domain module (the tasks app) using the CLI.