0
# Database & Models
1
2
SQLAlchemy-based database integration with automatic migrations, session management, async support, and built-in PostgreSQL optimization for modern web application data persistence.
3
4
## Capabilities
5
6
### Database Models
7
8
Base model class with built-in fields, relationship support, and automatic table creation for rapid application development.
9
10
```python { .api }
11
class Model(Base, sqlmodel.SQLModel):
12
"""
13
Base class to define a table in the database.
14
15
Inherits from both Reflex Base and SQLModel, providing automatic
16
table creation, relationship support, and seamless integration with
17
the Reflex framework and FastAPI/Pydantic ecosystem.
18
19
The default primary key field is automatically dropped if any other
20
field is marked as primary key.
21
"""
22
23
# The primary key for the table (auto-dropped if custom primary key exists)
24
id: int | None = sqlmodel.Field(default=None, primary_key=True)
25
26
def __init_subclass__(cls):
27
"""
28
Drop the default primary key field if any primary key field is defined.
29
30
Automatically handles custom primary key configurations by removing
31
the default 'id' field when other fields are marked as primary keys.
32
"""
33
...
34
35
@classmethod
36
def _dict_recursive(cls, value: Any):
37
"""
38
Recursively serialize the relationship object(s).
39
40
Args:
41
value: The value to serialize
42
43
Returns:
44
The serialized value
45
"""
46
...
47
48
def dict(self, **kwargs):
49
"""
50
Convert the object to a dictionary.
51
52
Includes both base fields and relationship data when available.
53
Handles SQLModel relationships that don't appear in __fields__.
54
55
Args:
56
kwargs: Ignored but needed for compatibility
57
58
Returns:
59
The object as a dictionary with fields and relationships
60
"""
61
...
62
63
class ModelRegistry:
64
"""
65
Registry for all models in the application.
66
67
Manages model registration and metadata collection for database
68
operations, migrations, and table creation.
69
"""
70
71
models: ClassVar[set[SQLModelOrSqlAlchemy]] = set()
72
73
@classmethod
74
def register(cls, model: SQLModelOrSqlAlchemy):
75
"""
76
Register a model. Can be used directly or as a decorator.
77
78
Args:
79
model: The model to register
80
81
Returns:
82
The model passed in as an argument (allows decorator usage)
83
"""
84
...
85
86
@classmethod
87
def get_models(cls, include_empty: bool = False) -> set[SQLModelOrSqlAlchemy]:
88
"""
89
Get registered models.
90
91
Args:
92
include_empty: If True, include models with empty metadata
93
94
Returns:
95
The registered models
96
"""
97
...
98
99
@classmethod
100
def get_metadata(cls) -> sqlalchemy.MetaData:
101
"""
102
Get the metadata for all registered models.
103
104
Returns:
105
Combined metadata from all registered models
106
"""
107
...
108
109
@classmethod
110
def filter(cls, **conditions) -> list[Model]:
111
"""
112
Filter model instances by field conditions.
113
114
Args:
115
**conditions: Field name to value mappings for filtering
116
117
Returns:
118
List of model instances matching conditions
119
"""
120
...
121
122
@classmethod
123
def count(cls) -> int:
124
"""
125
Count total number of instances in database.
126
127
Returns:
128
Total count of records for this model
129
"""
130
...
131
```
132
133
Usage example:
134
135
```python
136
import reflex as rx
137
from datetime import datetime
138
139
class User(rx.Model, table=True):
140
username: str
141
email: str
142
created_at: datetime = datetime.now()
143
is_active: bool = True
144
145
class Post(rx.Model, table=True):
146
title: str
147
content: str
148
author_id: int # Foreign key to User
149
published: bool = False
150
151
# Relationship
152
author: User = rx.Relationship(back_populates="posts")
153
154
# Usage
155
user = User.create(username="john", email="john@example.com")
156
post = Post.create(title="Hello World", content="My first post", author_id=user.id)
157
```
158
159
### Database Sessions
160
161
Session management for database operations with both synchronous and asynchronous support for optimal performance.
162
163
```python { .api }
164
def session() -> Session:
165
"""
166
Get database session for synchronous operations.
167
168
Provides SQLAlchemy session with automatic connection management,
169
transaction handling, and cleanup for database operations.
170
171
Returns:
172
SQLAlchemy Session instance for database operations
173
"""
174
...
175
176
def asession() -> AsyncSession:
177
"""
178
Get async database session for high-performance operations.
179
180
Provides async SQLAlchemy session for non-blocking database
181
operations, ideal for high-concurrency applications.
182
183
Returns:
184
SQLAlchemy AsyncSession for async database operations
185
"""
186
...
187
```
188
189
Usage examples:
190
191
```python
192
# Synchronous database operations
193
with rx.session() as db:
194
users = db.query(User).filter(User.is_active == True).all()
195
new_user = User(username="alice", email="alice@example.com")
196
db.add(new_user)
197
db.commit()
198
199
# Asynchronous database operations
200
async def get_users_async():
201
async with rx.asession() as db:
202
result = await db.execute(select(User).where(User.is_active == True))
203
return result.scalars().all()
204
```
205
206
### Database Configuration
207
208
Comprehensive database configuration with connection pooling, migration management, and PostgreSQL optimization.
209
210
```python { .api }
211
class DBConfig:
212
"""
213
Database configuration with PostgreSQL optimization.
214
215
Manages database connections, migration settings, and performance
216
tuning for production and development environments.
217
"""
218
219
db_url: str
220
"""Database connection URL (PostgreSQL, SQLite, MySQL, etc.)."""
221
222
engine: Engine | None = None
223
"""SQLAlchemy engine instance (auto-created if None)."""
224
225
migrate: bool = True
226
"""Whether to run automatic database migrations on startup."""
227
228
pool_size: int = 10
229
"""Number of connections to maintain in connection pool."""
230
231
max_overflow: int = 20
232
"""Maximum overflow connections beyond pool_size."""
233
234
pool_timeout: int = 30
235
"""Timeout in seconds for getting connection from pool."""
236
237
pool_recycle: int = 3600
238
"""Seconds before connection is recreated (prevents stale connections)."""
239
240
echo: bool = False
241
"""Whether to log all SQL statements (useful for debugging)."""
242
243
echo_pool: bool = False
244
"""Whether to log connection pool operations."""
245
246
def __init__(self, db_url: str, **kwargs) -> None:
247
"""
248
Initialize database configuration.
249
250
Args:
251
db_url: Database connection URL
252
**kwargs: Additional configuration options to override defaults
253
"""
254
...
255
256
def create_engine(self) -> Engine:
257
"""
258
Create SQLAlchemy engine with optimized settings.
259
260
Returns:
261
Configured SQLAlchemy Engine instance
262
"""
263
...
264
265
def create_tables(self) -> None:
266
"""
267
Create all database tables based on model definitions.
268
269
Uses SQLAlchemy metadata to create tables for all registered
270
models, handling dependencies and foreign key constraints.
271
"""
272
...
273
274
def run_migrations(self) -> None:
275
"""
276
Run database migrations to update schema.
277
278
Applies pending Alembic migrations to bring database schema
279
up to date with current model definitions.
280
"""
281
...
282
```
283
284
Database configuration in `rxconfig.py`:
285
286
```python
287
import reflex as rx
288
289
config = rx.Config(
290
app_name="myapp",
291
db_config=rx.DBConfig(
292
db_url="postgresql://user:pass@localhost/mydb",
293
pool_size=20,
294
max_overflow=30,
295
echo=False # Set to True for SQL debugging
296
)
297
)
298
```
299
300
### Query Building
301
302
Advanced query capabilities with relationship loading, filtering, and aggregation for complex data operations.
303
304
```python { .api }
305
# Model querying patterns
306
class QueryMixin:
307
"""Query building utilities for database models."""
308
309
@classmethod
310
def where(cls, *conditions) -> Query:
311
"""
312
Build query with WHERE conditions.
313
314
Args:
315
*conditions: SQLAlchemy filter conditions
316
317
Returns:
318
Query object with applied conditions
319
"""
320
...
321
322
@classmethod
323
def join(cls, *models) -> Query:
324
"""
325
Build query with JOIN operations.
326
327
Args:
328
*models: Model classes to join with
329
330
Returns:
331
Query object with joins applied
332
"""
333
...
334
335
@classmethod
336
def order_by(cls, *fields) -> Query:
337
"""
338
Build query with ORDER BY clause.
339
340
Args:
341
*fields: Fields to order by (use .desc() for descending)
342
343
Returns:
344
Query object with ordering applied
345
"""
346
...
347
348
@classmethod
349
def limit(cls, count: int) -> Query:
350
"""
351
Limit query results to specific count.
352
353
Args:
354
count: Maximum number of results to return
355
356
Returns:
357
Query object with limit applied
358
"""
359
...
360
361
@classmethod
362
def offset(cls, count: int) -> Query:
363
"""
364
Skip specified number of results.
365
366
Args:
367
count: Number of results to skip
368
369
Returns:
370
Query object with offset applied
371
"""
372
...
373
```
374
375
### Relationships
376
377
Model relationship definitions for foreign keys, one-to-many, and many-to-many associations with automatic loading.
378
379
```python { .api }
380
from sqlalchemy.orm import relationship
381
382
# Relationship patterns for models
383
class User(rx.Model, table=True):
384
id: int = Field(primary_key=True)
385
username: str
386
email: str
387
388
# One-to-many relationship
389
posts: list["Post"] = relationship("Post", back_populates="author")
390
profile: "UserProfile" = relationship("UserProfile", back_populates="user", uselist=False)
391
392
class Post(rx.Model, table=True):
393
id: int = Field(primary_key=True)
394
title: str
395
content: str
396
author_id: int = Field(foreign_key="user.id")
397
398
# Many-to-one relationship
399
author: User = relationship("User", back_populates="posts")
400
401
# Many-to-many relationship
402
tags: list["Tag"] = relationship("Tag", secondary="post_tags", back_populates="posts")
403
404
class Tag(rx.Model, table=True):
405
id: int = Field(primary_key=True)
406
name: str
407
408
# Many-to-many relationship
409
posts: list[Post] = relationship("Post", secondary="post_tags", back_populates="tags")
410
411
# Association table for many-to-many
412
class PostTag(rx.Model, table=True):
413
post_id: int = Field(foreign_key="post.id", primary_key=True)
414
tag_id: int = Field(foreign_key="tag.id", primary_key=True)
415
```
416
417
### Migration Management
418
419
Database schema versioning and migration utilities for evolving application data structures safely in production.
420
421
```python { .api }
422
def create_migration(message: str) -> None:
423
"""
424
Create new database migration file.
425
426
Generates Alembic migration file comparing current model
427
definitions against existing database schema.
428
429
Args:
430
message: Descriptive message for the migration
431
"""
432
...
433
434
def run_migrations() -> None:
435
"""
436
Apply pending migrations to database.
437
438
Runs all pending Alembic migrations to bring database
439
schema up to date with current model definitions.
440
"""
441
...
442
443
def rollback_migration(revision: str = None) -> None:
444
"""
445
Rollback database to previous migration.
446
447
Args:
448
revision: Specific revision to rollback to (latest if None)
449
"""
450
...
451
452
def get_migration_status() -> dict:
453
"""
454
Get current migration status and pending changes.
455
456
Returns:
457
Dictionary with current revision and pending migration info
458
"""
459
...
460
```
461
462
## Usage Examples
463
464
### Basic Model Definition
465
466
```python
467
import reflex as rx
468
from datetime import datetime
469
from typing import Optional
470
471
class User(rx.Model, table=True):
472
# Primary key (automatic)
473
id: int = Field(primary_key=True)
474
475
# Required fields
476
username: str = Field(unique=True, index=True)
477
email: str = Field(unique=True)
478
479
# Optional fields
480
full_name: Optional[str] = None
481
is_active: bool = True
482
483
# Timestamp fields (automatic)
484
created_at: datetime = Field(default_factory=datetime.utcnow)
485
updated_at: datetime = Field(default_factory=datetime.utcnow)
486
```
487
488
### Advanced Querying
489
490
```python
491
class UserState(rx.State):
492
users: list[User] = []
493
494
def load_active_users(self):
495
with rx.session() as db:
496
self.users = db.query(User).filter(
497
User.is_active == True
498
).order_by(User.created_at.desc()).all()
499
500
def search_users(self, query: str):
501
with rx.session() as db:
502
self.users = db.query(User).filter(
503
User.username.contains(query) |
504
User.full_name.contains(query)
505
).limit(10).all()
506
```
507
508
### Async Database Operations
509
510
```python
511
class AsyncUserState(rx.State):
512
users: list[User] = []
513
514
async def load_users_async(self):
515
async with rx.asession() as db:
516
result = await db.execute(
517
select(User).where(User.is_active == True)
518
.order_by(User.created_at.desc())
519
)
520
self.users = result.scalars().all()
521
```
522
523
## Types
524
525
```python { .api }
526
from typing import Any, Dict, List, Optional, Union
527
from sqlalchemy import Engine, Column
528
from sqlalchemy.orm import Session
529
from sqlalchemy.ext.asyncio import AsyncSession
530
from datetime import datetime
531
532
# Database Types
533
ModelInstance = Any # Database model instance
534
QueryResult = List[ModelInstance] # Query result list
535
ModelDict = Dict[str, Any] # Model as dictionary
536
537
# Session Types
538
DBSession = Session # Synchronous database session
539
AsyncDBSession = AsyncSession # Asynchronous database session
540
541
# Configuration Types
542
DatabaseURL = str # Database connection URL
543
EngineConfig = Dict[str, Any] # SQLAlchemy engine configuration
544
545
# Field Types
546
PrimaryKey = int # Primary key field type
547
ForeignKey = int # Foreign key field type
548
Timestamp = datetime # Timestamp field type
549
550
# Query Types
551
WhereCondition = Any # SQLAlchemy WHERE condition
552
JoinCondition = Any # SQLAlchemy JOIN condition
553
OrderByField = Any # SQLAlchemy ORDER BY field
554
555
# Migration Types
556
RevisionID = str # Alembic revision identifier
557
MigrationMessage = str # Migration description
558
```