refactor: rm unused function and added structured type

services/web/server/src/simcore_service_webserver/security_api.py

@@ -1,7 +1,6 @@
 """ API for security subsystem.
 
 """
-# pylint: disable=assignment-from-no-return
 import logging
 
 import passlib.hash
@@ -14,29 +13,13 @@
     is_anonymous,
     remember,
 )
-from aiopg.sa import Engine
 
-from .db_models import UserStatus, users
 from .security_authorization import AuthorizationPolicy, RoleBasedAccessModel
 from .security_roles import UserRole
 
 log = logging.getLogger(__name__)
 
 
-async def check_credentials(engine: Engine, email: str, password: str) -> bool:
-    async with engine.acquire() as conn:
-        query = users.select().where(
-            (users.c.email == email)
-            & (users.c.status != UserStatus.BANNED)
-            & (users.c.status != UserStatus.EXPIRED)
-        )
-        ret = await conn.execute(query)
-        user = await ret.fetchone()
-        if user is not None:
-            return check_password(password, user["password_hash"])
-    return False
-
-
 def encrypt_password(password: str) -> str:
     return passlib.hash.sha256_crypt.hash(password, rounds=1000)
 
@@ -55,14 +38,15 @@ def clean_auth_policy_cache(app: web.Application) -> None:
     autz_policy.timed_cache.clear()
 
 
-__all__ = (
-    "encrypt_password",
-    "check_credentials",
+__all__: tuple[str, ...] = (
     "authorized_userid",
-    "forget",
-    "remember",
-    "is_anonymous",
     "check_permission",
+    "encrypt_password",
+    "forget",
     "get_access_model",
+    "is_anonymous",
+    "remember",
     "UserRole",
 )
+
+# nopycln: file

services/web/server/src/simcore_service_webserver/security_authorization.py

@@ -1,14 +1,17 @@
 import logging
-from typing import Optional, Union
+from typing import Optional, TypedDict, Union
 
 import attr
+import sqlalchemy as sa
 from aiohttp import web
 from aiohttp_security.abc import AbstractAuthorizationPolicy
 from aiopg.sa import Engine
-from aiopg.sa.result import ResultProxy, RowProxy
+from aiopg.sa.result import ResultProxy
 from expiringdict import ExpiringDict
+from models_library.basic_types import IdInt
 from servicelib.aiohttp.aiopg_utils import PostgresRetryPolicyUponOperation
 from servicelib.aiohttp.application_keys import APP_DB_ENGINE_KEY
+from simcore_postgres_database.models.users import UserRole
 from tenacity import retry
 
 from .db_models import UserStatus, users
@@ -17,6 +20,11 @@
 log = logging.getLogger(__name__)
 
 
+class _UserIdentity(TypedDict, total=True):
+    id: IdInt
+    role: UserRole
+
+
 @attr.s(auto_attribs=True, frozen=True)
 class AuthorizationPolicy(AbstractAuthorizationPolicy):
     app: web.Application
@@ -36,22 +44,26 @@ def engine(self) -> Engine:
         return self.app[APP_DB_ENGINE_KEY]
 
     @retry(**PostgresRetryPolicyUponOperation(log).kwargs)
-    async def _pg_query_user(self, identity: str) -> Optional[RowProxy]:
+    async def _pg_query_user(self, identity: str) -> Optional[_UserIdentity]:
         # NOTE: Keeps a cache for a few seconds. Observed successive streams of this query
-        row = self.timed_cache.get(identity)
-        if row is None:
-            query = users.select().where(
-                (users.c.email == identity)
-                & (users.c.status != UserStatus.BANNED)
-                & (users.c.status != UserStatus.EXPIRED)
-            )
+        user: Optional[_UserIdentity] = self.timed_cache.get(identity)
+        if user is None:
             async with self.engine.acquire() as conn:
                 # NOTE: sometimes it raises psycopg2.DatabaseError in #880 and #1160
-                res: ResultProxy = await conn.execute(query)
-                row: Optional[RowProxy] = await res.fetchone()
+                result: ResultProxy = await conn.execute(
+                    sa.select([users.c.id, users.c.role]).where(
+                        (users.c.email == identity)
+                        & (users.c.status != UserStatus.BANNED)
+                        & (users.c.status != UserStatus.EXPIRED)
+                    )
+                )
+                row = await result.fetchone()
             if row is not None:
-                self.timed_cache[identity] = row
-        return row
+                assert row["id"]  # nosec
+                assert row["role"]  # nosec
+                self.timed_cache[identity] = dict(row.items())
+
+        return user
 
     async def authorized_userid(self, identity: str) -> Optional[int]:
         """Retrieve authorized user id.
@@ -60,7 +72,7 @@ async def authorized_userid(self, identity: str) -> Optional[int]:
         or "None" if no user exists related to the identity.
         """
         # TODO: why users.c.user_login_key!=users.c.email
-        user = await self._pg_query_user(identity)
+        user: Optional[_UserIdentity] = await self._pg_query_user(identity)
         return user["id"] if user else None
 
     async def permits(
@@ -78,9 +90,9 @@ async def permits(
         """
         if identity is None or permission is None:
             log.debug(
-                "Invalid indentity [%s] of permission [%s]. Denying access.",
-                identity,
-                permission,
+                "Invalid %s of %s. Denying access.",
+                f"{identity=}",
+                f"{permission=}",
             )
             return False