0
# Session Management
1

2
Session management with Flask application context integration, bind key resolution, and automatic session cleanup. Core database operations and transaction handling.
3

4
## Capabilities
5

6
### Session Class
7

8
Enhanced SQLAlchemy session with Flask application context integration and automatic bind selection.
9

10
```python { .api }
11
class Session(sa_orm.Session):
12
    """
13
    SQLAlchemy Session class with Flask-specific bind key resolution.
14
    
15
    Automatically selects the appropriate database engine based on
16
    the bind key associated with the metadata of queried models/tables.
17
    """
18
    
19
    def __init__(self, db: SQLAlchemy, **kwargs: Any) -> None:
20
        """
21
        Initialize session with reference to SQLAlchemy extension.
22
        
23
        Parameters:
24
        - db: SQLAlchemy extension instance
25
        - kwargs: Additional session arguments
26
        """
27
    
28
    def get_bind(
29
        self,
30
        mapper: Any | None = None,
31
        clause: Any | None = None,
32
        bind: sa.engine.Engine | sa.engine.Connection | None = None,
33
        **kwargs: Any,
34
    ) -> sa.engine.Engine | sa.engine.Connection:
35
        """
36
        Select database engine based on bind key of queried models/tables.
37
        
38
        Parameters:
39
        - mapper: Model mapper or class
40
        - clause: SQL clause or table
41
        - bind: Explicit bind to use (overrides automatic selection)
42
        - kwargs: Additional arguments
43
        
44
        Returns:
45
        Database engine or connection for the query
46
        """
47
```
48

49
### Session Access
50

51
The extension provides a scoped session accessible through the `session` property.
52

53
```python { .api }
54
@property
55
def session(self) -> sa_orm.scoped_session[Session]:
56
    """
57
    Scoped session factory for database operations.
58
    
59
    Sessions are scoped to Flask application context and automatically
60
    removed when the context exits. Requires active app context.
61
    """
62
```
63

64
### Query Convenience Methods
65

66
Extension methods that combine session operations with Flask error handling.
67

68
```python { .api }
69
def get_or_404(
70
    self,
71
    entity: type[Any],
72
    ident: Any,
73
    *,
74
    description: str | None = None,
75
    **kwargs: Any,
76
) -> Any:
77
    """
78
    Get entity by primary key or abort with 404 Not Found.
79
    
80
    Parameters:
81
    - entity: Model class to query
82
    - ident: Primary key value
83
    - description: Custom error message
84
    - kwargs: Additional arguments for session.get()
85
    
86
    Returns:
87
    Model instance
88
    
89
    Raises:
90
    404 error if entity not found
91
    """
92

93
def first_or_404(
94
    self, statement: sa.sql.Select[Any], *, description: str | None = None
95
) -> Any:
96
    """
97
    Execute select statement and return first result or abort with 404.
98
    
99
    Parameters:
100
    - statement: SQLAlchemy select statement
101
    - description: Custom error message
102
    
103
    Returns:
104
    First result from query
105
    
106
    Raises:
107
    404 error if no results found
108
    """
109

110
def one_or_404(
111
    self, statement: sa.sql.Select[Any], *, description: str | None = None
112
) -> Any:
113
    """
114
    Execute select statement expecting exactly one result or abort with 404.
115
    
116
    Parameters:
117
    - statement: SQLAlchemy select statement  
118
    - description: Custom error message
119
    
120
    Returns:
121
    Single result from query
122
    
123
    Raises:
124
    404 error if no results or multiple results found
125
    """
126
```
127

128
### Session Pagination
129

130
Built-in pagination support for select statements.
131

132
```python { .api }
133
def paginate(
134
    self,
135
    select: sa.sql.Select[Any],
136
    *,
137
    page: int | None = None,
138
    per_page: int | None = None,
139
    max_per_page: int | None = None,
140
    error_out: bool = True,
141
    count: bool = True,
142
) -> Pagination:
143
    """
144
    Apply pagination to a select statement.
145
    
146
    Parameters:
147
    - select: SQLAlchemy select statement to paginate
148
    - page: Current page number (from request args if None) 
149
    - per_page: Items per page (from request args if None)
150
    - max_per_page: Maximum allowed per_page value
151
    - error_out: Abort with 404 on invalid page parameters
152
    - count: Calculate total item count
153
    
154
    Returns:
155
    Pagination object with results and metadata
156
    """
157
```
158

159
### Utility Functions
160

161
Helper functions for session and bind management.
162

163
```python { .api }
164
def _clause_to_engine(
165
    clause: sa.ClauseElement | None,
166
    engines: Mapping[str | None, sa.engine.Engine],
167
) -> sa.engine.Engine | None:
168
    """
169
    Get engine from table metadata bind key.
170
    
171
    Parameters:
172
    - clause: SQL clause containing table reference
173
    - engines: Available engines by bind key
174
    
175
    Returns:
176
    Engine for the clause's bind key or None
177
    """
178

179
def _app_ctx_id() -> int:
180
    """
181
    Get Flask application context ID for session scoping.
182
    
183
    Returns:
184
    Unique identifier for current app context
185
    """
186
```
187

188
## Usage Examples
189

190
### Basic Session Usage
191

192
```python
193
# Session is automatically available and scoped to app context
194
with app.app_context():
195
    # Add new record
196
    user = User(username='john', email='john@example.com')
197
    db.session.add(user)
198
    db.session.commit()
199
    
200
    # Query records
201
    users = db.session.execute(db.select(User)).scalars().all()
202
```
203

204
### Using Query Convenience Methods
205

206
```python
207
@app.route('/user/<int:user_id>')
208
def get_user(user_id):
209
    # Automatically returns 404 if user not found
210
    user = db.get_or_404(User, user_id)
211
    return f"User: {user.username}"
212

213
@app.route('/users/first')
214
def first_user():
215
    # Get first user or 404 if none exist
216
    user = db.first_or_404(db.select(User).order_by(User.id))
217
    return f"First user: {user.username}"
218

219
@app.route('/user/unique/<username>')  
220
def unique_user(username):
221
    # Expect exactly one match or 404
222
    user = db.one_or_404(db.select(User).where(User.username == username))
223
    return f"User: {user.username}"
224
```
225

226
### Multiple Database Sessions
227

228
```python
229
# Session automatically selects correct engine based on model bind key
230
class User(db.Model):
231
    __bind_key__ = 'users'
232
    id = db.Column(db.Integer, primary_key=True)
233

234
class Log(db.Model):
235
    __bind_key__ = 'logs'
236
    id = db.Column(db.Integer, primary_key=True)
237

238
with app.app_context():
239
    # These use different databases automatically
240
    db.session.add(User(name='john'))  # Uses 'users' database
241
    db.session.add(Log(message='test'))  # Uses 'logs' database
242
    db.session.commit()
243
```
244

245
### Session Pagination
246

247
```python
248
@app.route('/users')
249
def list_users():
250
    # Automatically handles page and per_page from request args
251
    pagination = db.paginate(
252
        db.select(User).order_by(User.username),
253
        per_page=20,
254
        max_per_page=100
255
    )
256
    
257
    return {
258
        'users': [u.username for u in pagination.items],
259
        'page': pagination.page,
260
        'pages': pagination.pages,
261
        'total': pagination.total
262
    }
263
```
264

265
### Manual Transaction Control  
266

267
```python
268
with app.app_context():
269
    try:
270
        user = User(username='john')
271
        db.session.add(user)
272
        
273
        # Explicit flush to get ID without committing
274
        db.session.flush()
275
        user_id = user.id
276
        
277
        # More operations...
278
        db.session.commit()
279
    except Exception:
280
        db.session.rollback()
281
        raise
282
```
283

284
### Session Configuration
285

286
```python
287
# Custom session options during extension initialization
288
db = SQLAlchemy(
289
    app,
290
    session_options={
291
        'class_': CustomSession,  # Use custom session class
292
        'expire_on_commit': False,  # Don't expire objects after commit
293
        'autoflush': False,  # Manual flush control
294
    }
295
)
296
```
297

298
## Session Lifecycle
299

300
1. **Creation**: Session created when first accessed in app context
301
2. **Binding**: Automatic engine selection based on model/table bind keys  
302
3. **Operations**: Add, query, update, delete operations
303
4. **Cleanup**: Session automatically removed when app context exits
304
5. **Error Handling**: Session rollback on exceptions, cleanup on teardown