From 9a974bbcb9038e760360e21261e62c83aa73535e Mon Sep 17 00:00:00 2001 From: Giancarlo Romeo Date: Wed, 15 Jul 2026 15:10:42 +0200 Subject: [PATCH 1/7] optimize --- .../repository/groups.py | 26 +++++++++++-------- 1 file changed, 15 insertions(+), 11 deletions(-) diff --git a/services/catalog/src/simcore_service_catalog/repository/groups.py b/services/catalog/src/simcore_service_catalog/repository/groups.py index f7413f5770d0..944c7bc7d9a0 100644 --- a/services/catalog/src/simcore_service_catalog/repository/groups.py +++ b/services/catalog/src/simcore_service_catalog/repository/groups.py @@ -7,24 +7,28 @@ from pydantic.types import PositiveInt from simcore_postgres_database.models.groups import GroupType, groups, user_to_groups from simcore_postgres_database.models.users import users +from simcore_postgres_database.utils_repos import pass_or_acquire_connection +from sqlalchemy.ext.asyncio import AsyncConnection from ..errors import UninitializedGroupError from ._base import BaseRepository class GroupsRepository(BaseRepository): - async def list_user_groups(self, user_id: int) -> list[GroupAtDB]: - async with self.db_engine.connect() as conn: - return [ - GroupAtDB.model_validate(row) - async for row in await conn.stream( - sa.select(groups) - .select_from( - user_to_groups.join(groups, user_to_groups.c.gid == groups.c.gid), - ) - .where(user_to_groups.c.uid == user_id) + async def list_user_groups( + self, + user_id: int, + connection: AsyncConnection | None = None, + ) -> list[GroupAtDB]: + async with pass_or_acquire_connection(self.db_engine, connection) as conn: + result = await conn.execute( + sa.select(groups) + .select_from( + user_to_groups.join(groups, user_to_groups.c.gid == groups.c.gid), ) - ] + .where(user_to_groups.c.uid == user_id) + ) + return TypeAdapter(list[GroupAtDB]).validate_python(result.mappings().all()) async def get_everyone_group(self) -> GroupAtDB: async with self.db_engine.connect() as conn: From 907ee5cce8f1fb76a85c4d4ec633bb2747a8d977 Mon Sep 17 00:00:00 2001 From: Giancarlo Romeo Date: Wed, 15 Jul 2026 15:28:34 +0200 Subject: [PATCH 2/7] optimize --- .../repository/groups.py | 47 +++++++++++-------- 1 file changed, 28 insertions(+), 19 deletions(-) diff --git a/services/catalog/src/simcore_service_catalog/repository/groups.py b/services/catalog/src/simcore_service_catalog/repository/groups.py index 944c7bc7d9a0..dff359bdf29c 100644 --- a/services/catalog/src/simcore_service_catalog/repository/groups.py +++ b/services/catalog/src/simcore_service_catalog/repository/groups.py @@ -30,39 +30,48 @@ async def list_user_groups( ) return TypeAdapter(list[GroupAtDB]).validate_python(result.mappings().all()) - async def get_everyone_group(self) -> GroupAtDB: - async with self.db_engine.connect() as conn: + async def get_everyone_group( + self, + connection: AsyncConnection | None = None, + ) -> GroupAtDB: + async with pass_or_acquire_connection(self.db_engine, connection) as conn: result = await conn.execute(sa.select(groups).where(groups.c.type == GroupType.EVERYONE)) row = result.first() if not row: raise UninitializedGroupError(group=GroupType.EVERYONE, repo_cls=GroupsRepository) return GroupAtDB.model_validate(row) - async def get_user_gid_from_email(self, user_email: LowerCaseEmailStr) -> PositiveInt | None: - async with self.db_engine.connect() as conn: + async def get_user_gid_from_email( + self, + user_email: LowerCaseEmailStr, + connection: AsyncConnection | None = None, + ) -> PositiveInt | None: + async with pass_or_acquire_connection(self.db_engine, connection) as conn: return cast( PositiveInt | None, await conn.scalar(sa.select(users.c.primary_gid).where(users.c.email == user_email)), ) - async def get_gid_from_affiliation(self, affiliation: str) -> PositiveInt | None: - async with self.db_engine.connect() as conn: - return cast( - PositiveInt | None, - await conn.scalar(sa.select(groups.c.gid).where(groups.c.name == affiliation)), - ) - - async def get_user_email_from_gid(self, gid: PositiveInt) -> LowerCaseEmailStr | None: - async with self.db_engine.connect() as conn: - email = await conn.scalar(sa.select(users.c.email).where(users.c.primary_gid == gid)) - return email or None + async def get_user_email_from_gid( + self, + gid: PositiveInt, + connection: AsyncConnection | None = None, + ) -> LowerCaseEmailStr | None: + async with pass_or_acquire_connection(self.db_engine, connection) as conn: + result = await conn.scalar(sa.select(users.c.email).where(users.c.primary_gid == gid)) + return TypeAdapter(LowerCaseEmailStr).validate_python(result) if result else None - async def list_user_emails_from_gids(self, gids: set[PositiveInt]) -> dict[PositiveInt, LowerCaseEmailStr | None]: + async def list_user_emails_from_gids( + self, + gids: set[PositiveInt], + connection: AsyncConnection | None = None, + ) -> dict[PositiveInt, LowerCaseEmailStr | None]: service_owners: dict[PositiveInt, LowerCaseEmailStr | None] = {} - async with self.db_engine.connect() as conn: - async for row in await conn.stream( + async with pass_or_acquire_connection(self.db_engine, connection) as conn: + result = await conn.execute( sa.select(users.c.primary_gid, users.c.email).where(users.c.primary_gid.in_(gids)) - ): + ) + for row in result: service_owners[row.primary_gid] = ( TypeAdapter(LowerCaseEmailStr).validate_python(row.email) if row.email else None ) From 29183f7b311180473473893cc792245eb14cd4bb Mon Sep 17 00:00:00 2001 From: Giancarlo Romeo Date: Wed, 15 Jul 2026 15:30:45 +0200 Subject: [PATCH 3/7] optimize --- .../src/simcore_service_catalog/repository/groups.py | 8 ++------ 1 file changed, 2 insertions(+), 6 deletions(-) diff --git a/services/catalog/src/simcore_service_catalog/repository/groups.py b/services/catalog/src/simcore_service_catalog/repository/groups.py index dff359bdf29c..0702e49910f6 100644 --- a/services/catalog/src/simcore_service_catalog/repository/groups.py +++ b/services/catalog/src/simcore_service_catalog/repository/groups.py @@ -1,5 +1,3 @@ -from typing import cast - import sqlalchemy as sa from models_library.emails import LowerCaseEmailStr from models_library.groups import GroupAtDB @@ -47,10 +45,8 @@ async def get_user_gid_from_email( connection: AsyncConnection | None = None, ) -> PositiveInt | None: async with pass_or_acquire_connection(self.db_engine, connection) as conn: - return cast( - PositiveInt | None, - await conn.scalar(sa.select(users.c.primary_gid).where(users.c.email == user_email)), - ) + result = await conn.scalar(sa.select(users.c.primary_gid).where(users.c.email == user_email)) + return TypeAdapter(PositiveInt).validate_python(result) if result else None async def get_user_email_from_gid( self, From 786febdbe509f02a31dbf0ff9ce2e1dee8fc0d01 Mon Sep 17 00:00:00 2001 From: Giancarlo Romeo Date: Tue, 11 Aug 2026 17:49:51 +0200 Subject: [PATCH 4/7] optimize --- .../src/simcore_service_catalog/repository/groups.py | 9 +++------ 1 file changed, 3 insertions(+), 6 deletions(-) diff --git a/services/catalog/src/simcore_service_catalog/repository/groups.py b/services/catalog/src/simcore_service_catalog/repository/groups.py index 489a3ef4972b..c1c09753f626 100644 --- a/services/catalog/src/simcore_service_catalog/repository/groups.py +++ b/services/catalog/src/simcore_service_catalog/repository/groups.py @@ -58,13 +58,10 @@ async def list_user_emails_from_gids( gids: set[PositiveInt], connection: AsyncConnection | None = None, ) -> dict[PositiveInt, LowerCaseEmailStr | None]: - service_owners: dict[PositiveInt, LowerCaseEmailStr | None] = {} async with pass_or_acquire_connection(self.db_engine, connection) as conn: result = await conn.execute( sa.select(users.c.primary_gid, users.c.email).where(users.c.primary_gid.in_(gids)) ) - for row in result: - service_owners[row.primary_gid] = ( - TypeAdapter(LowerCaseEmailStr).validate_python(row.email) if row.email else None - ) - return service_owners + return TypeAdapter(dict[PositiveInt, LowerCaseEmailStr | None]).validate_python( + {row.primary_gid: row.email for row in result} + ) From 00f92ae65a6fe422fd9cf7ce897fb25ca77957a3 Mon Sep 17 00:00:00 2001 From: Giancarlo Romeo Date: Fri, 28 Aug 2026 09:43:01 +0200 Subject: [PATCH 5/7] use connections --- .../repository/services.py | 29 +++++++++------ .../service/catalog_services.py | 37 ++++++++++++------- 2 files changed, 40 insertions(+), 26 deletions(-) diff --git a/services/catalog/src/simcore_service_catalog/repository/services.py b/services/catalog/src/simcore_service_catalog/repository/services.py index eeba4387aee9..1feb36f388b4 100644 --- a/services/catalog/src/simcore_service_catalog/repository/services.py +++ b/services/catalog/src/simcore_service_catalog/repository/services.py @@ -32,6 +32,7 @@ from sqlalchemy import sql from sqlalchemy.dialects.postgresql import insert as pg_insert from sqlalchemy.exc import IntegrityError +from sqlalchemy.ext.asyncio import AsyncConnection from ..models.services_db import ( ReleaseDBGet, @@ -256,9 +257,10 @@ async def can_get_service( # get args key: ServiceKey, version: ServiceVersion, + connection: AsyncConnection | None = None, ) -> bool: """Returns False if it cannot get the service i.e. not found or does not have access""" - async with self.db_engine.begin() as conn: + async with pass_or_acquire_connection(self.db_engine, connection) as conn: result = await conn.execute( _services_sql.can_get_service_stmt( product_name=product_name, @@ -278,8 +280,9 @@ async def can_update_service( # get args key: ServiceKey, version: ServiceVersion, + connection: AsyncConnection | None = None, ) -> bool: - async with self.db_engine.begin() as conn: + async with pass_or_acquire_connection(self.db_engine, connection) as conn: result = await conn.execute( _services_sql.can_get_service_stmt( product_name=product_name, @@ -299,6 +302,7 @@ async def get_service_with_history( # get args key: ServiceKey, version: ServiceVersion, + connection: AsyncConnection | None = None, ) -> ServiceWithHistoryDBGet | None: stmt_get = _services_sql.get_service_stmt( product_name=product_name, @@ -308,21 +312,21 @@ async def get_service_with_history( service_version=version, ) - async with self.db_engine.begin() as conn: + async with pass_or_acquire_connection(self.db_engine, connection) as conn: result = await conn.execute(stmt_get) row = result.one_or_none() - if row: - stmt_history = _services_sql.get_service_history_stmt( - product_name=product_name, - user_id=user_id, - access_rights=AccessRightsClauses.can_read, - service_key=key, - ) - async with self.db_engine.begin() as conn: + if row: + stmt_history = _services_sql.get_service_history_stmt( + product_name=product_name, + user_id=user_id, + access_rights=AccessRightsClauses.can_read, + service_key=key, + ) result = await conn.execute(stmt_history) row_h = result.one_or_none() + if row: return ServiceWithHistoryDBGet( key=row.key, version=row.version, @@ -597,6 +601,7 @@ async def get_service_access_rights( key: str, version: str, product_name: str | None = None, + connection: AsyncConnection | None = None, ) -> list[ServiceAccessRightsDB]: """ - If product_name is not specified, then all are considered in the query @@ -607,7 +612,7 @@ async def get_service_access_rights( query = sa.select(services_access_rights).where(search_expression) - async with self.db_engine.connect() as conn: + async with pass_or_acquire_connection(self.db_engine, connection) as conn: return [ServiceAccessRightsDB.model_validate(row) async for row in await conn.stream(query)] async def batch_get_services_access_rights_or_none( diff --git a/services/catalog/src/simcore_service_catalog/service/catalog_services.py b/services/catalog/src/simcore_service_catalog/service/catalog_services.py index a07355569b15..d8858e00e4a3 100644 --- a/services/catalog/src/simcore_service_catalog/service/catalog_services.py +++ b/services/catalog/src/simcore_service_catalog/service/catalog_services.py @@ -30,6 +30,8 @@ CatalogInconsistentRpcError, CatalogItemNotFoundRpcError, ) +from simcore_postgres_database.utils_repos import pass_or_acquire_connection +from sqlalchemy.ext.asyncio import AsyncConnection from ..clients.director import DirectorClient from ..errors import BatchNotFoundError @@ -383,21 +385,24 @@ async def get_catalog_service( service_key: ServiceKey, service_version: ServiceVersion, ) -> ServiceGetV2: - access_rights = await check_catalog_service_permissions( - repo=repo, - product_name=product_name, - user_id=user_id, - service_key=service_key, - service_version=service_version, - permission="read", - ) + async with pass_or_acquire_connection(repo.db_engine) as connection: + access_rights = await check_catalog_service_permissions( + repo=repo, + product_name=product_name, + user_id=user_id, + service_key=service_key, + service_version=service_version, + permission="read", + connection=connection, + ) - service = await repo.get_service_with_history( - product_name=product_name, - user_id=user_id, - key=service_key, - version=service_version, - ) + service = await repo.get_service_with_history( + product_name=product_name, + user_id=user_id, + key=service_key, + version=service_version, + connection=connection, + ) if not service: # no service found provided `access_rights` raise CatalogForbiddenRpcError( @@ -512,6 +517,7 @@ async def check_catalog_service_permissions( service_key: ServiceKey, service_version: ServiceVersion, permission: Literal["read", "write"], + connection: AsyncConnection | None = None, ) -> list[ServiceAccessRightsDB]: """Raises if the service cannot be accessed with the specified permission level @@ -532,6 +538,7 @@ async def check_catalog_service_permissions( key=service_key, version=service_version, product_name=product_name, + connection=connection, ) if not access_rights: raise CatalogItemNotFoundRpcError( @@ -549,6 +556,7 @@ async def check_catalog_service_permissions( user_id=user_id, key=service_key, version=service_version, + connection=connection, ) elif permission == "write": has_permission = await repo.can_update_service( @@ -556,6 +564,7 @@ async def check_catalog_service_permissions( user_id=user_id, key=service_key, version=service_version, + connection=connection, ) if not has_permission: From c3d4aa354f6cf597a6c72c73e172f0e057d256df Mon Sep 17 00:00:00 2001 From: Giancarlo Romeo Date: Fri, 28 Aug 2026 09:50:18 +0200 Subject: [PATCH 6/7] remove stream --- .../catalog/src/simcore_service_catalog/repository/services.py | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/services/catalog/src/simcore_service_catalog/repository/services.py b/services/catalog/src/simcore_service_catalog/repository/services.py index 1feb36f388b4..6815ae399868 100644 --- a/services/catalog/src/simcore_service_catalog/repository/services.py +++ b/services/catalog/src/simcore_service_catalog/repository/services.py @@ -613,7 +613,8 @@ async def get_service_access_rights( query = sa.select(services_access_rights).where(search_expression) async with pass_or_acquire_connection(self.db_engine, connection) as conn: - return [ServiceAccessRightsDB.model_validate(row) async for row in await conn.stream(query)] + result = await conn.execute(query) + return [ServiceAccessRightsDB.model_validate(row) for row in result] async def batch_get_services_access_rights_or_none( self, From 8d2b217416a350fa478b76091ffc05fee39a9f65 Mon Sep 17 00:00:00 2001 From: Giancarlo Romeo Date: Fri, 28 Aug 2026 09:53:23 +0200 Subject: [PATCH 7/7] group --- .../service/catalog_services.py | 45 +++++++++++++------ 1 file changed, 32 insertions(+), 13 deletions(-) diff --git a/services/catalog/src/simcore_service_catalog/service/catalog_services.py b/services/catalog/src/simcore_service_catalog/service/catalog_services.py index d8858e00e4a3..358f2cc095a0 100644 --- a/services/catalog/src/simcore_service_catalog/service/catalog_services.py +++ b/services/catalog/src/simcore_service_catalog/service/catalog_services.py @@ -377,6 +377,32 @@ async def list_latest_catalog_services( return total_count, items +async def _get_service_access_rights_or_raise( + repo: ServicesRepository, + *, + product_name: ProductName, + user_id: UserID, + service_key: ServiceKey, + service_version: ServiceVersion, + connection: AsyncConnection | None = None, +) -> list[ServiceAccessRightsDB]: + access_rights = await repo.get_service_access_rights( + key=service_key, + version=service_version, + product_name=product_name, + connection=connection, + ) + if not access_rights: + raise CatalogItemNotFoundRpcError( + name=f"{service_key}:{service_version}", + service_key=service_key, + service_version=service_version, + user_id=user_id, + product_name=product_name, + ) + return access_rights + + async def get_catalog_service( repo: ServicesRepository, director_api: DirectorClient, @@ -386,13 +412,12 @@ async def get_catalog_service( service_version: ServiceVersion, ) -> ServiceGetV2: async with pass_or_acquire_connection(repo.db_engine) as connection: - access_rights = await check_catalog_service_permissions( + access_rights = await _get_service_access_rights_or_raise( repo=repo, product_name=product_name, user_id=user_id, service_key=service_key, service_version=service_version, - permission="read", connection=connection, ) @@ -534,20 +559,14 @@ async def check_catalog_service_permissions( CatalogForbiddenError: insufficient access rights to get the requested access """ - access_rights = await repo.get_service_access_rights( - key=service_key, - version=service_version, + access_rights = await _get_service_access_rights_or_raise( + repo=repo, product_name=product_name, + user_id=user_id, + service_key=service_key, + service_version=service_version, connection=connection, ) - if not access_rights: - raise CatalogItemNotFoundRpcError( - name=f"{service_key}:{service_version}", - service_key=service_key, - service_version=service_version, - user_id=user_id, - product_name=product_name, - ) has_permission = False if permission == "read":