params = dict(compiled.params)
if isinstance(extra_params, Mapping):
for k, v in extra_params.items():
- params[k] = str(v)
+ if hasattr(v, 'to_wkt'):
+ params[k] = v.to_wkt()
+ elif isinstance(v, (int, float)):
+ params[k] = v
+ else:
+ params[k] = str(v)
elif isinstance(extra_params, Sequence) and extra_params:
for k in extra_params[0]:
params[k] = f':{k}'
sqlstr = str(compiled)
- if sa.__version__.startswith('1'):
- try:
- sqlstr = re.sub(r'__\[POSTCOMPILE_[^]]*\]', '%s', sqlstr)
- return sqlstr % tuple((repr(params.get(name, None))
- for name in compiled.positiontup)) # type: ignore
- except TypeError:
- return sqlstr
-
- # Fixes an odd issue with Python 3.7 where percentages are not
- # quoted correctly.
- sqlstr = re.sub(r'%(?!\()', '%%', sqlstr)
- sqlstr = re.sub(r'__\[POSTCOMPILE_([^]]*)\]', r'%(\1)s', sqlstr)
- print(sqlstr)
- return sqlstr % params
+ if conn.dialect.name == 'postgresql':
+ if sa.__version__.startswith('1'):
+ try:
+ sqlstr = re.sub(r'__\[POSTCOMPILE_[^]]*\]', '%s', sqlstr)
+ return sqlstr % tuple((repr(params.get(name, None))
+ for name in compiled.positiontup)) # type: ignore
+ except TypeError:
+ return sqlstr
+
+ # Fixes an odd issue with Python 3.7 where percentages are not
+ # quoted correctly.
+ sqlstr = re.sub(r'%(?!\()', '%%', sqlstr)
+ sqlstr = re.sub(r'__\[POSTCOMPILE_([^]]*)\]', r'%(\1)s', sqlstr)
+ return sqlstr % params
+
+ assert conn.dialect.name == 'sqlite'
+
+ # params in positional order
+ pparams = (repr(params.get(name, None)) for name in compiled.positiontup) # type: ignore
+
+ sqlstr = re.sub(r'__\[POSTCOMPILE_([^]]*)\]', '?', sqlstr)
+ sqlstr = re.sub(r"\?", lambda m: next(pparams), sqlstr)
+
+ return sqlstr
class HTMLLogger(BaseLogger):
""" Logger that formats messages in HTML.
self.buffer = io.StringIO()
+ def _timestamp(self) -> None:
+ self._write(f'[{dt.datetime.now()}]\n')
+
+
def get_buffer(self) -> str:
return self.buffer.getvalue()
def section(self, heading: str) -> None:
+ self._timestamp()
self._write(f"\n# {heading}\n\n")
def result_dump(self, heading: str, results: Iterator[Tuple[Any, Any]]) -> None:
+ self._timestamp()
self._write(f'{heading}:\n')
total = 0
for rank, res in results:
def sql(self, conn: AsyncConnection, statement: 'sa.Executable',
params: Union[Mapping[str, Any], Sequence[Mapping[str, Any]], None]) -> None:
+ self._timestamp()
sqlstr = '\n| '.join(textwrap.wrap(self.format_sql(conn, statement, params), width=78))
self._write(f"| {sqlstr}\n\n")