0
# Session Management
1

2
SQLModel provides enhanced database session classes with improved type safety for executing SQLModel queries. These sessions extend SQLAlchemy's session functionality with better integration for SQLModel's type system.
3

4
## Capabilities
5

6
### Synchronous Session
7

8
Enhanced SQLAlchemy Session with better typing support for SQLModel operations.
9

10
```python { .api }
11
class Session(SQLAlchemySession):
12
    """
13
    Enhanced SQLAlchemy Session with improved typing for SQLModel.
14
    
15
    Provides the same functionality as SQLAlchemy Session but with better
16
    type inference when executing SQLModel select statements.
17
    """
18
    
19
    def exec(
20
        self,
21
        statement: Select[_TSelectParam],
22
        *,
23
        params: Optional[Union[Mapping[str, Any], Sequence[Mapping[str, Any]]]] = None,
24
        execution_options: Mapping[str, Any] = {},
25
        bind_arguments: Optional[Dict[str, Any]] = None,
26
        _parent_execute_state: Optional[Any] = None,
27
        _add_event: Optional[Any] = None,
28
    ) -> TupleResult[_TSelectParam]:
29
        """
30
        Execute a select statement and return results with proper typing.
31
        
32
        Parameters:
33
            statement: SQLModel Select statement to execute
34
            params: Parameters for the statement
35
            execution_options: Execution-specific options
36
            bind_arguments: Bind-specific arguments
37
            
38
        Returns:
39
            TupleResult for regular selects, ScalarResult for scalar selects
40
        """
41
    
42
    def exec(
43
        self,
44
        statement: SelectOfScalar[_TSelectParam],
45
        *,
46
        params: Optional[Union[Mapping[str, Any], Sequence[Mapping[str, Any]]]] = None,
47
        execution_options: Mapping[str, Any] = {},
48
        bind_arguments: Optional[Dict[str, Any]] = None,
49
        _parent_execute_state: Optional[Any] = None,
50
        _add_event: Optional[Any] = None,
51
    ) -> ScalarResult[_TSelectParam]:
52
        """Execute a scalar select statement."""
53
    
54
    def exec(
55
        self,
56
        statement: Executable,
57
        *,
58
        params: Optional[Union[Mapping[str, Any], Sequence[Mapping[str, Any]]]] = None,
59
        execution_options: Mapping[str, Any] = {},
60
        bind_arguments: Optional[Dict[str, Any]] = None,
61
        _parent_execute_state: Optional[Any] = None,
62
        _add_event: Optional[Any] = None,
63
    ) -> Result[Any]:
64
        """Execute any other executable statement."""
65
    
66
    def execute(
67
        self,
68
        statement: Executable,
69
        params: Optional[Any] = None,
70
        *,
71
        execution_options: Any = {},
72
        bind_arguments: Optional[Dict[str, Any]] = None,
73
        _parent_execute_state: Optional[Any] = None,
74
        _add_event: Optional[Any] = None,
75
    ) -> Result[Any]:
76
        """
77
        🚨 You probably want to use `session.exec()` instead of `session.execute()`.
78

79
        This is the original SQLAlchemy `session.execute()` method that returns objects
80
        of type `Row`, and that you have to call `scalars()` to get the model objects.
81
        """
82
    
83
    def query(self, *entities: Any, **kwargs: Any) -> Any:
84
        """
85
        🚨 You probably want to use `session.exec()` instead of `session.query()`.
86

87
        `session.exec()` is SQLModel's own short version with increased type
88
        annotations.
89
        """
90
```
91

92
**Usage Examples:**
93
```python
94
from sqlmodel import Session, create_engine, select
95

96
# Create engine and session
97
engine = create_engine("sqlite:///database.db")
98

99
# Basic query execution
100
with Session(engine) as session:
101
    # Type-safe query execution
102
    statement = select(Hero).where(Hero.name == "Spider-Boy")
103
    result = session.exec(statement)
104
    hero = result.first()  # Returns Hero | None with proper typing
105
    
106
    # Scalar queries
107
    count_statement = select(func.count(Hero.id))
108
    count_result = session.exec(count_statement)
109
    count = count_result.one()  # Returns int with proper typing
110

111
# Transaction management
112
with Session(engine) as session:
113
    # Create new record
114
    hero = Hero(name="New Hero", secret_name="Secret")
115
    session.add(hero)
116
    
117
    # Update existing record
118
    statement = select(Hero).where(Hero.id == 1)
119
    existing_hero = session.exec(statement).first()
120
    if existing_hero:
121
        existing_hero.age = 25
122
        session.add(existing_hero)
123
    
124
    # Commit all changes
125
    session.commit()
126
    
127
    # Session automatically closes and rolls back on exception
128
```
129

130

131
### Session Integration with SQLModel
132

133
The Session class is designed to work seamlessly with SQLModel's type system and provide enhanced developer experience through better typing and integration.
134

135
**Key Benefits:**
136

137
1. **Type Safety**: The `exec()` method provides proper return types based on the query type
138
2. **Better IDE Support**: Full autocompletion and type checking for query results
139
3. **Consistent API**: Same interface as SQLAlchemy sessions but with SQLModel enhancements
140
4. **Backward Compatibility**: Can be used as drop-in replacements for SQLAlchemy sessions
141

142
**Common Patterns:**
143
```python
144
# Repository pattern with type safety
145
class HeroRepository:
146
    def __init__(self, session: Session):
147
        self.session = session
148
    
149
    def get_by_id(self, hero_id: int) -> Optional[Hero]:
150
        statement = select(Hero).where(Hero.id == hero_id)
151
        result = self.session.exec(statement)
152
        return result.first()  # Typed as Optional[Hero]
153
    
154
    def get_all(self) -> List[Hero]:
155
        statement = select(Hero)
156
        result = self.session.exec(statement)
157
        return result.all()  # Typed as List[Hero]
158
    
159
    def create(self, hero: Hero) -> Hero:
160
        self.session.add(hero)
161
        self.session.commit()
162
        self.session.refresh(hero)
163
        return hero
164

165
# Dependency injection (FastAPI example)
166
def get_session():
167
    with Session(engine) as session:
168
        yield session
169

170
@app.get("/heroes/{hero_id}")
171
def get_hero(hero_id: int, session: Session = Depends(get_session)) -> Hero:
172
    statement = select(Hero).where(Hero.id == hero_id)
173
    result = session.exec(statement)
174
    hero = result.first()
175
    if not hero:
176
        raise HTTPException(status_code=404, detail="Hero not found")
177
    return hero  # Automatically serialized as JSON by FastAPI
178
```