Coverage for sqlmodel/orm/session.py: 100%

1from typing import ( 6 ctx1fabcde

2 Any, 

3 Dict, 

4 Mapping, 

5 Optional, 

6 Sequence, 

7 TypeVar, 

8 Union, 

9 overload, 

10) 

11 

12from sqlalchemy import util 6 ctx1fabcde

13from sqlalchemy.engine.interfaces import _CoreAnyExecuteParams 6 ctx1fabcde

14from sqlalchemy.engine.result import Result, ScalarResult, TupleResult 6 ctx1fabcde

15from sqlalchemy.orm import Query as _Query 6 ctx1fabcde

16from sqlalchemy.orm import Session as _Session 6 ctx1fabcde

17from sqlalchemy.orm._typing import OrmExecuteOptionsParameter 6 ctx1fabcde

18from sqlalchemy.sql._typing import _ColumnsClauseArgument 6 ctx1fabcde

19from sqlalchemy.sql.base import Executable as _Executable 6 ctx1fabcde

20from sqlmodel.sql.base import Executable 6 ctx1fabcde

21from sqlmodel.sql.expression import Select, SelectOfScalar 6 ctx1fabcde

22from typing_extensions import deprecated 6 ctx1fabcde

23 

24_TSelectParam = TypeVar("_TSelectParam", bound=Any) 6 ctx1fabcde

25 

26 

27class Session(_Session): 6 ctx1fabcde

28 @overload 6 ctx1fabcde

29 def exec( 5 ctx1abcde

30 self, 

31 statement: Select[_TSelectParam], 5 ctx1abcde

32 *, 

33 params: Optional[Union[Mapping[str, Any], Sequence[Mapping[str, Any]]]] = None, 6 ctx1fabcde

34 execution_options: Mapping[str, Any] = util.EMPTY_DICT, 6 ctx1fabcde

35 bind_arguments: Optional[Dict[str, Any]] = None, 6 ctx1fabcde

36 _parent_execute_state: Optional[Any] = None, 6 ctx1fabcde

37 _add_event: Optional[Any] = None, 6 ctx1fabcde

38 ) -> TupleResult[_TSelectParam]: ... 6 ctx1fabcde

39 

40 @overload 6 ctx1fabcde

41 def exec( 5 ctx1abcde

42 self, 

43 statement: SelectOfScalar[_TSelectParam], 5 ctx1abcde

44 *, 

45 params: Optional[Union[Mapping[str, Any], Sequence[Mapping[str, Any]]]] = None, 6 ctx1fabcde

46 execution_options: Mapping[str, Any] = util.EMPTY_DICT, 6 ctx1fabcde

47 bind_arguments: Optional[Dict[str, Any]] = None, 6 ctx1fabcde

48 _parent_execute_state: Optional[Any] = None, 6 ctx1fabcde

49 _add_event: Optional[Any] = None, 6 ctx1fabcde

50 ) -> ScalarResult[_TSelectParam]: ... 6 ctx1fabcde

51 

52 def exec( 5 ctx1abcde

53 self, 

54 statement: Union[ 

55 Select[_TSelectParam], 

56 SelectOfScalar[_TSelectParam], 

57 Executable[_TSelectParam], 

58 ], 

59 *, 

60 params: Optional[Union[Mapping[str, Any], Sequence[Mapping[str, Any]]]] = None, 

61 execution_options: Mapping[str, Any] = util.EMPTY_DICT, 

62 bind_arguments: Optional[Dict[str, Any]] = None, 

63 _parent_execute_state: Optional[Any] = None, 

64 _add_event: Optional[Any] = None, 

65 ) -> Union[TupleResult[_TSelectParam], ScalarResult[_TSelectParam]]: 

66 results = super().execute( 6 ctx1fabcde

67 statement, 

68 params=params, 

69 execution_options=execution_options, 

70 bind_arguments=bind_arguments, 

71 _parent_execute_state=_parent_execute_state, 

72 _add_event=_add_event, 

73 ) 

74 if isinstance(statement, SelectOfScalar): 6 ctx1fabcde

75 return results.scalars() 6 ctx1fabcde

76 return results # type: ignore 6 ctx1fabcde

77 

78 @deprecated( 6 ctx1fabcde

79 """ 

80 🚨 You probably want to use `session.exec()` instead of `session.execute()`. 

81 

82 This is the original SQLAlchemy `session.execute()` method that returns objects 

83 of type `Row`, and that you have to call `scalars()` to get the model objects. 

84 

85 For example: 

86 

87 ```Python 

88 heroes = session.execute(select(Hero)).scalars().all() 

89 ``` 

90 

91 instead you could use `exec()`: 

92 

93 ```Python 

94 heroes = session.exec(select(Hero)).all() 

95 ``` 

96 """, 

97 category=None, 

98 ) 

99 def execute( # type: ignore 5 ctx1abcde

100 self, 

101 statement: _Executable, 

102 params: Optional[_CoreAnyExecuteParams] = None, 

103 *, 

104 execution_options: OrmExecuteOptionsParameter = util.EMPTY_DICT, 

105 bind_arguments: Optional[Dict[str, Any]] = None, 

106 _parent_execute_state: Optional[Any] = None, 

107 _add_event: Optional[Any] = None, 

108 ) -> Result[Any]: 

109 """ 

110 🚨 You probably want to use `session.exec()` instead of `session.execute()`. 

111 

112 This is the original SQLAlchemy `session.execute()` method that returns objects 

113 of type `Row`, and that you have to call `scalars()` to get the model objects. 

114 

115 For example: 

116 

117 ```Python 

118 heroes = session.execute(select(Hero)).scalars().all() 

119 ``` 

120 

121 instead you could use `exec()`: 

122 

123 ```Python 

124 heroes = session.exec(select(Hero)).all() 

125 ``` 

126 """ 

127 return super().execute( 6 ctx1fabcde

128 statement, 

129 params=params, 

130 execution_options=execution_options, 

131 bind_arguments=bind_arguments, 

132 _parent_execute_state=_parent_execute_state, 

133 _add_event=_add_event, 

134 ) 

135 

136 @deprecated( 6 ctx1fabcde

137 """ 

138 🚨 You probably want to use `session.exec()` instead of `session.query()`. 

139 

140 `session.exec()` is SQLModel's own short version with increased type 

141 annotations. 

142 

143 Or otherwise you might want to use `session.execute()` instead of 

144 `session.query()`. 

145 """ 

146 ) 

147 def query( # type: ignore 5 ctx1abcde

148 self, *entities: _ColumnsClauseArgument[Any], **kwargs: Any 

149 ) -> _Query[Any]: 

150 """ 

151 🚨 You probably want to use `session.exec()` instead of `session.query()`. 

152 

153 `session.exec()` is SQLModel's own short version with increased type 

154 annotations. 

155 

156 Or otherwise you might want to use `session.execute()` instead of 

157 `session.query()`. 

158 """ 

159 return super().query(*entities, **kwargs) 6 ctx1fabcde