Use Concatenate to annotate do_execute (#12666)

This commit is contained in:
David Robertson 2022-05-09 10:28:38 +01:00 committed by GitHub
parent 0ce2201932
commit 26c1ad71c5
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
3 changed files with 16 additions and 6 deletions

1
changelog.d/12666.misc Normal file
View file

@ -0,0 +1 @@
Use `Concatenate` to better annotate `_do_execute`.

View file

@ -142,7 +142,7 @@ netaddr = ">=0.7.18"
# add a lower bound to the Jinja2 dependency.
Jinja2 = ">=3.0"
bleach = ">=1.4.3"
# We use `ParamSpec`, which was added in `typing-extensions` 3.10.0.0.
# We use `ParamSpec` and `Concatenate`, which were added in `typing-extensions` 3.10.0.0.
typing-extensions = ">=3.10.0"
# We enforce that we have a `cryptography` version that bundles an `openssl`
# with the latest security patches.

View file

@ -38,7 +38,7 @@ from typing import (
import attr
from prometheus_client import Histogram
from typing_extensions import Literal
from typing_extensions import Concatenate, Literal, ParamSpec
from twisted.enterprise import adbapi
@ -194,7 +194,7 @@ class LoggingDatabaseConnection:
# The type of entry which goes on our after_callbacks and exception_callbacks lists.
_CallbackListEntry = Tuple[Callable[..., object], Iterable[Any], Dict[str, Any]]
P = ParamSpec("P")
R = TypeVar("R")
@ -339,7 +339,13 @@ class LoggingTransaction:
"Strip newlines out of SQL so that the loggers in the DB are on one line"
return " ".join(line.strip() for line in sql.splitlines() if line.strip())
def _do_execute(self, func: Callable[..., R], sql: str, *args: Any) -> R:
def _do_execute(
self,
func: Callable[Concatenate[str, P], R],
sql: str,
*args: P.args,
**kwargs: P.kwargs,
) -> R:
sql = self._make_sql_one_line(sql)
# TODO(paul): Maybe use 'info' and 'debug' for values?
@ -348,7 +354,10 @@ class LoggingTransaction:
sql = self.database_engine.convert_param_style(sql)
if args:
try:
sql_logger.debug("[SQL values] {%s} %r", self.name, args[0])
# The type-ignore should be redundant once mypy releases a version with
# https://github.com/python/mypy/pull/12668. (`args` might be empty,
# (but we'll catch the index error if so.)
sql_logger.debug("[SQL values] {%s} %r", self.name, args[0]) # type: ignore[index]
except Exception:
# Don't let logging failures stop SQL from working
pass
@ -363,7 +372,7 @@ class LoggingTransaction:
opentracing.tags.DATABASE_STATEMENT: sql,
},
):
return func(sql, *args)
return func(sql, *args, **kwargs)
except Exception as e:
sql_logger.debug("[SQL FAIL] {%s} %s", self.name, e)
raise