3
3
import warnings
4
4
from pathlib import Path
5
5
from re import Match
6
- from typing import Any , Dict , List , Optional , Union
6
+ from typing import Any , Optional , Union
7
7
8
8
import antlr4 .tree .Tree
9
9
import sqlalchemy
@@ -28,7 +28,7 @@ def _code(path: Union[str, Path], encoding: str) -> str:
28
28
return "\n " .join (fh .readlines ())
29
29
30
30
31
- def _process_replacement (line : str , parameters : Dict [str , Any ]) -> str :
31
+ def _process_replacement (line : str , parameters : dict [str , Any ]) -> str :
32
32
"""Appropriately replace a single <replace> statement."""
33
33
new_line = line .format (** parameters )
34
34
if None in parameters .values ():
@@ -42,7 +42,7 @@ def _process_replacement(line: str, parameters: Dict[str, Any]) -> str:
42
42
43
43
def _parametrize (
44
44
source : str ,
45
- parameters : Dict [str , Any ],
45
+ parameters : dict [str , Any ],
46
46
start : str = _REPLACE_START ,
47
47
end : str = _REPLACE_END ,
48
48
) -> str :
@@ -93,7 +93,7 @@ def __init__(self):
93
93
# session information is lost after the execution of a batch.
94
94
# We therefore need to manually prepend it to all following
95
95
# batches.
96
- self .dynamics : List [str ] = []
96
+ self .dynamics : list [str ] = []
97
97
98
98
def visit (
99
99
self , tree : TSqlParser .Sql_clausesContext , prepend_dynamics : bool = True
@@ -112,7 +112,7 @@ def visit(
112
112
113
113
return " " .join (dynamics + chunks )
114
114
115
- def visitChildren (self , node : antlr4 .ParserRuleContext ) -> List [str ]: # noqa: N802
115
+ def visitChildren (self , node : antlr4 .ParserRuleContext ) -> list [str ]: # noqa: N802
116
116
if isinstance (node , TSqlParser .Print_statementContext ):
117
117
# Print statements are replaced by inserts into a temporary table so that they can be evaluated
118
118
# at the right time and fetched afterwards.
@@ -129,15 +129,15 @@ def visitChildren(self, node: antlr4.ParserRuleContext) -> List[str]: # noqa: N
129
129
130
130
return result
131
131
132
- def visitTerminal (self , node : antlr4 .TerminalNode ) -> List [str ]: # noqa: N802
132
+ def visitTerminal (self , node : antlr4 .TerminalNode ) -> list [str ]: # noqa: N802
133
133
return [str (node )]
134
134
135
- def defaultResult (self ) -> List [str ]: # noqa: N802
135
+ def defaultResult (self ) -> list [str ]: # noqa: N802
136
136
return []
137
137
138
138
def aggregateResult ( # noqa: N802
139
- self , aggregate : List [str ], next_result : List [str ]
140
- ) -> List [str ]:
139
+ self , aggregate : list [str ], next_result : list [str ]
140
+ ) -> list [str ]:
141
141
return aggregate + next_result
142
142
143
143
@@ -158,7 +158,7 @@ def syntaxError( # noqa: N802
158
158
raise ValueError (f"Error parsing SQL script: { error_message } " )
159
159
160
160
161
- def _split (code : str , isolate_top_level_statements : bool = True ) -> List [str ]:
161
+ def _split (code : str , isolate_top_level_statements : bool = True ) -> list [str ]:
162
162
if not USE_CPP_IMPLEMENTATION :
163
163
warnings .warn (
164
164
"Can not find C++ version of the parser, Python version will be used instead."
@@ -215,7 +215,7 @@ def _fetch_and_clear_prints(conn: Connection):
215
215
def executes (
216
216
code : str ,
217
217
engine : sqlalchemy .engine .Engine ,
218
- parameters : Optional [Dict [str , Any ]] = None ,
218
+ parameters : Optional [dict [str , Any ]] = None ,
219
219
isolate_top_level_statements = True ,
220
220
) -> None :
221
221
"""Execute a given sql string through a sqlalchemy.engine.Engine connection.
@@ -250,7 +250,7 @@ def executes(
250
250
def execute (
251
251
path : Union [str , Path ],
252
252
engine : sqlalchemy .engine .Engine ,
253
- parameters : Optional [Dict [str , Any ]] = None ,
253
+ parameters : Optional [dict [str , Any ]] = None ,
254
254
isolate_top_level_statements = True ,
255
255
encoding : str = "utf-8" ,
256
256
) -> None :
0 commit comments