Database Module

The Database module wraps SQLAlchemy async with intelligent query planning, N+1 query detection, automatic relationship loading, and performance modes. It provides a high-level repository pattern for common operations.

Configuration

example.py

python

Copied!

1	from vorte import Vorte
2
3	app = Vorte(
4	auto_load=True,
5	config={
6	"database": {
7	"url": "postgresql+asyncpg://user:pass@localhost:5432/mydb",
8	"pool_size": 20,
9	"max_overflow": 10,
10	"pool_timeout": 30,
11	"echo": False,
12	"performance_mode": "balanced",
13	"n_plus_1_detection": True,
14	},
15	},
16	)

Performance Modes

Mode	Pooling	Prepared Statements	N+1 Detection
`development`	Small pool	Disabled	Enabled with warnings
`balanced`	Medium pool	Enabled	Enabled with logging
`production`	Large pool	Enabled	Disabled

Basic Usage

list_users.py

python

Copied!

1	from vorte.database import Database, Repository
2
3	db = Database()
4
5	@router.get("/users")
6	async def list_users():
7	users = await db.fetch_all("SELECT * FROM users WHERE active = true")
8	return success_response(data=users)
9
10	@router.get("/users/{user_id}")
11	async def get_user(user_id: int):
12	user = await db.fetch_one(
13	"SELECT * FROM users WHERE id = :id",
14	{"id": user_id},
15	)
16	if not user:
17	return error_response(code="NOT_FOUND", status_code=404)
18	return success_response(data=user)

Repository Pattern

base.py

python

Copied!

1	from vorte.database import Repository
2	from sqlalchemy import Column, Integer, String, DateTime
3	from sqlalchemy.orm import DeclarativeBase
4
5	class Base(DeclarativeBase):
6	pass
7
8	class User(Base):
9	__tablename__ = "users"
10	id = Column(Integer, primary_key=True)
11	name = Column(String(255))
12	email = Column(String(255), unique=True)
13	created_at = Column(DateTime)
14
15	class UserRepository(Repository[User]):
16	model = User
17
18	async def find_by_email(self, email: str) -> User \| None:
19	return await self.find_one(User.email == email)
20
21	async def find_active(self, page: int = 1, per_page: int = 20):
22	return await self.paginate(
23	User.id > 0,
24	page=page,
25	per_page=per_page,
26	order_by=User.created_at.desc(),
27	)
28
29	user_repo = UserRepository()
30
31	@router.get("/users")
32	async def list_users(page: int = 1):
33	result = await user_repo.find_active(page=page)
34	return paginated_response(
35	items=result.items,
36	total=result.total,
37	page=page,
38	per_page=20,
39	)

N+1 Query Detection

The database module monitors queries within a request and detects N+1 patterns. When enabled, it logs warnings with the offending query sequence and suggests eager loading strategies.

schema.sql

sql

Copied!

1	# N+1 Detection is enabled in development and balanced modes
2	# It logs warnings when it detects the pattern:
3
4	# WARNING: N+1 query detected on /api/users
5	# Query 1: SELECT * FROM users LIMIT 20
6	# Queries 2-21: SELECT * FROM posts WHERE user_id = ?
7	# Suggestion: Use select_related(User.posts) or joinedload(User.posts)

select_related and prefetch

list_users_with_posts.py

python

Copied!

1	from vorte.database import select_related
2
3	@router.get("/users")
4	async def list_users_with_posts():
5	users = await db.fetch_all(
6	select_related(User, User.posts, User.profile)
7	.where(User.active == true())
8	.limit(20)
9	)
10	return success_response(data=users)

Query Planning

Vorte analyzes query patterns and generates optimized execution plans. It caches frequently used query plans and suggests indexes.

example.py

python

Copied!

1	from vorte.database import QueryPlanner
2
3	planner = QueryPlanner()
4
5	plan = await planner.analyze(
6	"SELECT u.*, COUNT(p.id) FROM users u JOIN posts p ON u.id = p.user_id GROUP BY u.id"
7	)
8
9	print(plan.estimated_cost)
10	print(plan.suggested_indexes)
11	print(plan.execution_time_ms)

Migrations

terminal

bash

Copied!

1	# Generate a migration
2	vorte db migrate --message "Add user preferences table"
3
4	# Apply migrations
5	vorte db upgrade
6
7	# Rollback last migration
8	vorte db downgrade
9
10	# Show migration history
11	vorte db history

Transactions

create_order.py

python

Copied!

1	from vorte.database import Database
2
3	db = Database()
4
5	@router.post("/orders")
6	async def create_order(payload: OrderPayload):
7	async with db.transaction():
8	order = await db.execute(
9	"INSERT INTO orders (user_id, total) VALUES (:uid, :total) RETURNING id",
10	{"uid": payload.user_id, "total": payload.total},
11	)
12	for item in payload.items:
13	await db.execute(
14	"INSERT INTO order_items (order_id, product_id, qty) VALUES (:oid, :pid, :qty)",
15	{"oid": order.id, "pid": item.product_id, "qty": item.qty},
16	)
17	return success_response(data={"order_id": order.id})

Database Module

Configuration

Performance Modes

Basic Usage

Repository Pattern

N+1 Query Detection

select_related and prefetch

Query Planning

Migrations

Transactions

Stay in the loop