0
# Database Engine and Connection
1

2
SQLModel re-exports SQLAlchemy's database engine and connection management functionality, providing full compatibility with SQLAlchemy's engine system while maintaining seamless integration with SQLModel's enhanced features.
3

4
## Capabilities
5

6
### Engine Creation
7

8
Functions for creating database engines that handle connection pooling and database communication.
9

10
```python { .api }
11
def create_engine(
12
    url: Union[str, URL],
13
    *,
14
    connect_args: Optional[Dict[str, Any]] = None,
15
    echo: bool = False,
16
    echo_pool: bool = False,
17
    enable_from_linting: bool = True,
18
    hide_parameters: bool = False,
19
    insertmanyvalues_page_size: int = 1000,
20
    isolation_level: Optional[str] = None,
21
    json_deserializer: Optional[Callable] = None,
22
    json_serializer: Optional[Callable] = None,
23
    label_length: Optional[int] = None,
24
    logging_name: Optional[str] = None,
25
    max_identifier_length: Optional[int] = None,
26
    max_overflow: int = 10,
27
    module: Optional[Any] = None,
28
    paramstyle: Optional[str] = None,
29
    pool: Optional[Pool] = None,
30
    poolclass: Optional[Type[Pool]] = None,
31
    pool_logging_name: Optional[str] = None,
32
    pool_pre_ping: bool = False,
33
    pool_recycle: int = -1,
34
    pool_reset_on_return: Optional[str] = None,
35
    pool_size: int = 5,
36
    pool_timeout: float = 30.0,
37
    plugins: Optional[List[str]] = None,
38
    query_cache_size: int = 500,
39
    **kwargs: Any
40
) -> Engine:
41
    """
42
    Create a database engine for connecting to databases.
43
    
44
    Parameters:
45
        url: Database URL (e.g., "sqlite:///database.db", "postgresql://user:pass@host/db")
46
        connect_args: Additional arguments passed to the database driver
47
        echo: Whether to log all SQL statements (useful for debugging)
48
        echo_pool: Whether to log connection pool events
49
        pool_size: Number of connections to maintain in the pool
50
        max_overflow: Additional connections beyond pool_size
51
        pool_timeout: Seconds to wait for connection from pool
52
        pool_recycle: Seconds after which connection is recreated
53
        pool_pre_ping: Validate connections before use
54
        isolation_level: Transaction isolation level
55
        
56
    Returns:
57
        Engine object for database operations
58
    """
59

60
def create_mock_engine(dialect: str, executor: Callable) -> MockConnection:
61
    """
62
    Create a mock engine for testing without a real database.
63
    
64
    Parameters:
65
        dialect: SQL dialect to mock
66
        executor: Function to handle SQL execution
67
        
68
    Returns:
69
        MockConnection for testing purposes
70
    """
71

72
def engine_from_config(
73
    configuration: Dict[str, Any],
74
    prefix: str = "sqlalchemy.",
75
    **kwargs: Any
76
) -> Engine:
77
    """
78
    Create an engine from a configuration dictionary.
79
    
80
    Parameters:
81
        configuration: Dictionary with engine configuration
82
        prefix: Prefix for configuration keys
83
        **kwargs: Additional engine arguments
84
        
85
    Returns:
86
        Engine configured from the dictionary
87
    """
88
```
89

90
**Usage Examples:**
91
```python
92
# SQLite engine (file-based)
93
engine = create_engine("sqlite:///heroes.db", echo=True)
94

95
# PostgreSQL engine
96
engine = create_engine(
97
    "postgresql://user:password@localhost/heroes",
98
    pool_size=20,
99
    max_overflow=0,
100
    pool_recycle=3600
101
)
102

103
# MySQL engine with connection arguments
104
engine = create_engine(
105
    "mysql+pymysql://user:password@localhost/heroes",
106
    connect_args={"charset": "utf8mb4"},
107
    echo=False
108
)
109

110
# In-memory SQLite for testing
111
engine = create_engine("sqlite:///:memory:", echo=True)
112

113
# Engine from configuration
114
config = {
115
    "sqlalchemy.url": "sqlite:///heroes.db",
116
    "sqlalchemy.echo": "true",
117
    "sqlalchemy.pool_size": "10"
118
}
119
engine = engine_from_config(config)
120
```
121

122
### Connection Pooling
123

124
Connection pool classes for managing database connections efficiently.
125

126
```python { .api }
127
class QueuePool(Pool):
128
    """
129
    Queue-based connection pool (default for most databases).
130
    
131
    Maintains a fixed-size pool of connections with overflow capability.
132
    Connections are distributed on a FIFO basis.
133
    """
134

135
class StaticPool(Pool):
136
    """
137
    Static connection pool that maintains a single connection.
138
    
139
    Useful for in-memory databases or single-connection scenarios.
140
    All operations share the same connection.
141
    """
142
```
143

144
**Usage Examples:**
145
```python
146
# Explicit pool configuration
147
from sqlalchemy.pool import QueuePool, StaticPool
148

149
# Custom queue pool
150
engine = create_engine(
151
    "postgresql://user:pass@host/db",
152
    poolclass=QueuePool,
153
    pool_size=10,
154
    max_overflow=20,
155
    pool_pre_ping=True
156
)
157

158
# Static pool for SQLite
159
engine = create_engine(
160
    "sqlite:///database.db",
161
    poolclass=StaticPool,
162
    connect_args={"check_same_thread": False}
163
)
164
```
165

166
### Database Inspection
167

168
Utility for inspecting database schemas and metadata.
169

170
```python { .api }
171
def inspect(bind: Union[Engine, Connection]) -> Inspector:
172
    """
173
    Create an Inspector for examining database structure.
174
    
175
    Parameters:
176
        bind: Engine or Connection to inspect
177
        
178
    Returns:
179
        Inspector object for schema reflection
180
    """
181
```
182

183
**Usage Examples:**
184
```python
185
# Inspect database schema
186
inspector = inspect(engine)
187

188
# Get table names
189
table_names = inspector.get_table_names()
190
print("Tables:", table_names)
191

192
# Get column information
193
columns = inspector.get_columns("heroes")
194
for column in columns:
195
    print(f"Column: {column['name']}, Type: {column['type']}")
196

197
# Get foreign keys
198
foreign_keys = inspector.get_foreign_keys("heroes")
199
for fk in foreign_keys:
200
    print(f"FK: {fk['constrained_columns']} -> {fk['referred_table']}.{fk['referred_columns']}")
201

202
# Check if table exists
203
if inspector.has_table("heroes"):
204
    print("Heroes table exists")
205
```
206

207
### Integration with SQLModel
208

209
The engine system integrates seamlessly with SQLModel's session management and model definitions.
210

211
**Complete Application Example:**
212
```python
213
from sqlmodel import SQLModel, Field, Session, create_engine, select
214
from typing import Optional
215

216
# Define model
217
class Hero(SQLModel, table=True):
218
    id: Optional[int] = Field(default=None, primary_key=True)
219
    name: str
220
    secret_name: str
221
    age: Optional[int] = None
222

223
# Create engine
224
engine = create_engine("sqlite:///heroes.db", echo=True)
225

226
# Create tables
227
SQLModel.metadata.create_all(engine)
228

229
# Use with sessions
230
def create_hero(name: str, secret_name: str, age: Optional[int] = None) -> Hero:
231
    hero = Hero(name=name, secret_name=secret_name, age=age)
232
    with Session(engine) as session:
233
        session.add(hero)
234
        session.commit()
235
        session.refresh(hero)
236
        return hero
237

238
def get_heroes() -> List[Hero]:
239
    with Session(engine) as session:
240
        statement = select(Hero)
241
        heroes = session.exec(statement).all()
242
        return heroes
243

244
# Usage
245
hero = create_hero("Spider-Boy", "Pedro Parqueador", 16)
246
all_heroes = get_heroes()
247
```
248

249
### Engine Configuration Best Practices
250

251
**Production Configuration:**
252
```python
253
# Production PostgreSQL setup
254
engine = create_engine(
255
    "postgresql://user:password@host:5432/database",
256
    # Connection pool settings
257
    pool_size=20,          # Base number of connections
258
    max_overflow=30,       # Additional connections when needed
259
    pool_recycle=3600,     # Recreate connections after 1 hour
260
    pool_pre_ping=True,    # Validate connections before use
261
    
262
    # Performance settings
263
    echo=False,            # Disable SQL logging
264
    query_cache_size=1000, # Increase query cache
265
    
266
    # Connection arguments
267
    connect_args={
268
        "sslmode": "require",
269
        "connect_timeout": 10,
270
        "application_name": "my_app"
271
    }
272
)
273
```
274

275
**Development Configuration:**
276
```python
277
# Development SQLite setup
278
engine = create_engine(
279
    "sqlite:///dev_database.db",
280
    echo=True,  # Enable SQL logging for debugging
281
    connect_args={"check_same_thread": False}  # Allow multiple threads
282
)
283
```
284

285
**Testing Configuration:**
286
```python
287
# In-memory database for tests
288
test_engine = create_engine(
289
    "sqlite:///:memory:",
290
    echo=False,
291
    poolclass=StaticPool,
292
    connect_args={"check_same_thread": False}
293
)
294
```