aboutsummaryrefslogtreecommitdiff
path: root/gn3/auth
diff options
context:
space:
mode:
Diffstat (limited to 'gn3/auth')
-rw-r--r--gn3/auth/authentication/oauth2/endpoints/introspection.py4
-rw-r--r--gn3/auth/authentication/oauth2/endpoints/utilities.py4
-rw-r--r--gn3/auth/authentication/oauth2/grants/authorisation_code_grant.py2
-rw-r--r--gn3/auth/authentication/oauth2/models/oauth2token.py4
-rw-r--r--gn3/auth/db.py8
5 files changed, 9 insertions, 13 deletions
diff --git a/gn3/auth/authentication/oauth2/endpoints/introspection.py b/gn3/auth/authentication/oauth2/endpoints/introspection.py
index a567363..29ade6a 100644
--- a/gn3/auth/authentication/oauth2/endpoints/introspection.py
+++ b/gn3/auth/authentication/oauth2/endpoints/introspection.py
@@ -24,7 +24,7 @@ class IntrospectionEndpoint(_IntrospectionEndpoint):
"""Query the token."""
return _query_token(self, token_string, token_type_hint)
- def introspect_token(self, token: OAuth2Token) -> dict:# pylint: disable=[no-self-use]
+ def introspect_token(self, token: OAuth2Token) -> dict:
"""Return the introspection information."""
url = urlparse(flask_request.url)
return {
@@ -43,6 +43,6 @@ class IntrospectionEndpoint(_IntrospectionEndpoint):
"jti": token.token_id
}
- def check_permission(self, token, client, request):# pylint: disable=[unused-argument, no-self-use]
+ def check_permission(self, token, client, request):# pylint: disable=[unused-argument]
"""Check that the client has permission to introspect token."""
return client.client_type == "internal"
diff --git a/gn3/auth/authentication/oauth2/endpoints/utilities.py b/gn3/auth/authentication/oauth2/endpoints/utilities.py
index 299f151..b85940d 100644
--- a/gn3/auth/authentication/oauth2/endpoints/utilities.py
+++ b/gn3/auth/authentication/oauth2/endpoints/utilities.py
@@ -12,7 +12,9 @@ def query_token(# pylint: disable=[unused-argument]
endpoint_object: Any, token_str: str, token_type_hint) -> Optional[
OAuth2Token]:
"""Retrieve the token from the database."""
- __identity__ = lambda val: val
+ def __identity__(val):
+ """Identity function."""
+ return val
token = Nothing
with db.connection(current_app.config["AUTH_DB"]) as conn:
if token_type_hint == "access_token":
diff --git a/gn3/auth/authentication/oauth2/grants/authorisation_code_grant.py b/gn3/auth/authentication/oauth2/grants/authorisation_code_grant.py
index f80d02e..fb8d436 100644
--- a/gn3/auth/authentication/oauth2/grants/authorisation_code_grant.py
+++ b/gn3/auth/authentication/oauth2/grants/authorisation_code_grant.py
@@ -41,7 +41,7 @@ class AuthorisationCodeGrant(grants.AuthorizationCodeGrant):
"""Retrieve the code from the database."""
return __query_authorization_code__(code, client)
- def delete_authorization_code(self, authorization_code):# pylint: disable=[no-self-use]
+ def delete_authorization_code(self, authorization_code):
"""Delete the authorisation code."""
with db.connection(app.config["AUTH_DB"]) as conn:
with db.cursor(conn) as cursor:
diff --git a/gn3/auth/authentication/oauth2/models/oauth2token.py b/gn3/auth/authentication/oauth2/models/oauth2token.py
index 72e20cc..981e730 100644
--- a/gn3/auth/authentication/oauth2/models/oauth2token.py
+++ b/gn3/auth/authentication/oauth2/models/oauth2token.py
@@ -51,7 +51,9 @@ class OAuth2Token(NamedTuple):
return self.revoked
def __token_from_resultset__(conn: db.DbConnection, rset) -> Maybe:
- __identity__ = lambda val: val
+ def __identity__(val):
+ """Identity function."""
+ return val
try:
the_user = user_by_id(conn, uuid.UUID(rset["user_id"]))
except NotFoundError as _nfe:
diff --git a/gn3/auth/db.py b/gn3/auth/db.py
index 2ba6619..97f3982 100644
--- a/gn3/auth/db.py
+++ b/gn3/auth/db.py
@@ -10,40 +10,32 @@ class DbConnection(Protocol):
"""Type annotation for a generic database connection object."""
def cursor(self) -> Any:
"""A cursor object"""
- ...
def commit(self) -> Any:
"""Commit the transaction."""
- ...
def rollback(self) -> Any:
"""Rollback the transaction."""
- ...
class DbCursor(Protocol):
"""Type annotation for a generic database cursor object."""
def execute(self, *args, **kwargs) -> Any:
"""Execute a single query"""
- ...
def executemany(self, *args, **kwargs) -> Any:
"""
Execute parameterized SQL statement sql against all parameter sequences
or mappings found in the sequence parameters.
"""
- ...
def fetchone(self, *args, **kwargs):
"""Fetch single result if present, or `None`."""
- ...
def fetchmany(self, *args, **kwargs):
"""Fetch many results if present or `None`."""
- ...
def fetchall(self, *args, **kwargs):
"""Fetch all results if present or `None`."""
- ...
@contextlib.contextmanager
def connection(db_path: str, row_factory: Callable = sqlite3.Row) -> Iterator[DbConnection]: