11288123511

Committed 11 Oct 2024 07:23AM UTC coverage: 90.667% (+0.2%) from 90.477%

Build # 11288123511

Build Type

Pull #407

github

Committed by

web-flow

Commit Message

Merge 20c6c8af6 into 5b095d795

Pull Request Pull Request #407: feat!: add data connectors

Run Details

1226 of 1325 new or added lines in 28 files covered. (92.53%)

3 existing lines in 3 files now uncovered.

10589 of 11679 relevant lines covered (90.67%)

1.6 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

53.45

/bases/renku_data_services/background_jobs/config.py

"""Configurations for background jobs."""

import os
from collections.abc import Callable
from dataclasses import dataclass

from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine
from sqlalchemy.orm import sessionmaker

from renku_data_services.authz.authz import Authz
from renku_data_services.authz.config import AuthzConfig
from renku_data_services.data_connectors.db import DataConnectorProjectLinkRepository, DataConnectorRepository
from renku_data_services.data_connectors.migration_utils import DataConnectorMigrationTool
from renku_data_services.errors import errors
from renku_data_services.message_queue.config import RedisConfig
from renku_data_services.message_queue.db import EventRepository
from renku_data_services.message_queue.redis_queue import RedisQueue
from renku_data_services.namespace.db import GroupRepository
from renku_data_services.project.db import ProjectRepository
from renku_data_services.users.db import UserRepo, UsersSync
from renku_data_services.users.kc_api import IKeycloakAPI, KeycloakAPI


@dataclass
class SyncConfig:
    """Main configuration."""

    syncer: UsersSync
    kc_api: IKeycloakAPI
    authz_config: AuthzConfig
    group_repo: GroupRepository
    event_repo: EventRepository
    project_repo: ProjectRepository
    data_connector_migration_tool: DataConnectorMigrationTool
    session_maker: Callable[..., AsyncSession]

    @classmethod
    def from_env(cls, prefix: str = "") -> "SyncConfig":
        """Generate a configuration from environment variables."""
        pg_host = os.environ.get(f"{prefix}DB_HOST", "localhost")
        pg_user = os.environ.get(f"{prefix}DB_USER", "renku")
        pg_port = os.environ.get(f"{prefix}DB_PORT", "5432")
        db_name = os.environ.get(f"{prefix}DB_NAME", "renku")
        pg_password = os.environ.get(f"{prefix}DB_PASSWORD")
        if pg_password is None:
            raise errors.ConfigurationError(
                message="Please provide a database password in the 'DB_PASSWORD' environment variable."
            )
        async_sqlalchemy_url = f"postgresql+asyncpg://{pg_user}:{pg_password}@{pg_host}:{pg_port}/{db_name}"
        # NOTE: the pool here is not used to serve HTTP requests, it is only used in background jobs.
        # Therefore, we want to consume very few connections and we can wait for an available connection
        # much longer than the default 30 seconds. In our tests syncing 15 users times out with the default.
        engine = create_async_engine(async_sqlalchemy_url, pool_size=4, max_overflow=0, pool_timeout=600)
        session_maker: Callable[..., AsyncSession] = sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)  # type: ignore[call-overload]
        redis = RedisConfig.from_env(prefix)
        message_queue = RedisQueue(redis)

        authz_config = AuthzConfig.from_env()
        event_repo = EventRepository(session_maker=session_maker, message_queue=message_queue)
        group_repo = GroupRepository(
            session_maker,
            event_repo=event_repo,
            group_authz=Authz(authz_config),
            message_queue=message_queue,
        )
        project_repo = ProjectRepository(
            session_maker=session_maker,
            message_queue=message_queue,
            event_repo=event_repo,
            group_repo=group_repo,
            authz=Authz(authz_config),
        )
        data_connector_repo = DataConnectorRepository(
            session_maker=session_maker,
            authz=Authz(authz_config),
        )
        data_connector_project_link_repo = DataConnectorProjectLinkRepository(
            session_maker=session_maker,
            authz=Authz(authz_config),
        )
        data_connector_migration_tool = DataConnectorMigrationTool(
            session_maker=session_maker,
            data_connector_repo=data_connector_repo,
            data_connector_project_link_repo=data_connector_project_link_repo,
            project_repo=project_repo,
            authz=Authz(authz_config),
        )
        user_repo = UserRepo(
            session_maker=session_maker,
            message_queue=message_queue,
            event_repo=event_repo,
            group_repo=group_repo,
            encryption_key=None,
            authz=Authz(authz_config),
        )
        syncer = UsersSync(
            session_maker,
            message_queue=message_queue,
            event_repo=event_repo,
            group_repo=group_repo,
            user_repo=user_repo,
            authz=Authz(authz_config),
        )
        keycloak_url = os.environ[f"{prefix}KEYCLOAK_URL"]
        client_id = os.environ[f"{prefix}KEYCLOAK_CLIENT_ID"]
        client_secret = os.environ[f"{prefix}KEYCLOAK_CLIENT_SECRET"]
        realm = os.environ.get(f"{prefix}KEYCLOAK_REALM", "Renku")
        kc_api = KeycloakAPI(keycloak_url=keycloak_url, client_id=client_id, client_secret=client_secret, realm=realm)
        return cls(
            syncer,
            kc_api,
            authz_config,
            group_repo,
            event_repo,
            project_repo,
            data_connector_migration_tool,
            session_maker,
        )

1	"""Configurations for background jobs."""	2✔
2
3	import os	2✔
4	from collections.abc import Callable	2✔
5	from dataclasses import dataclass	2✔
6
7	from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine	2✔
8	from sqlalchemy.orm import sessionmaker	2✔
9
10	from renku_data_services.authz.authz import Authz	2✔
11	from renku_data_services.authz.config import AuthzConfig	2✔
12	from renku_data_services.data_connectors.db import DataConnectorProjectLinkRepository, DataConnectorRepository	2✔
13	from renku_data_services.data_connectors.migration_utils import DataConnectorMigrationTool	2✔
14	from renku_data_services.errors import errors	2✔
15	from renku_data_services.message_queue.config import RedisConfig	2✔
16	from renku_data_services.message_queue.db import EventRepository	2✔
17	from renku_data_services.message_queue.redis_queue import RedisQueue	2✔
18	from renku_data_services.namespace.db import GroupRepository	2✔
19	from renku_data_services.project.db import ProjectRepository	2✔
20	from renku_data_services.users.db import UserRepo, UsersSync	2✔
21	from renku_data_services.users.kc_api import IKeycloakAPI, KeycloakAPI	2✔
22
23
24	@dataclass	2✔
25	class SyncConfig:	2✔
26	"""Main configuration."""	2✔
27
28	syncer: UsersSync	2✔
29	kc_api: IKeycloakAPI	2✔
30	authz_config: AuthzConfig	2✔
31	group_repo: GroupRepository	2✔
32	event_repo: EventRepository	2✔
33	project_repo: ProjectRepository	2✔
34	data_connector_migration_tool: DataConnectorMigrationTool	2✔
35	session_maker: Callable[..., AsyncSession]	2✔
36
37	@classmethod	2✔
38	def from_env(cls, prefix: str = "") -> "SyncConfig":	2✔
39	"""Generate a configuration from environment variables."""
40	pg_host = os.environ.get(f"{prefix}DB_HOST", "localhost")	×
41	pg_user = os.environ.get(f"{prefix}DB_USER", "renku")	×
42	pg_port = os.environ.get(f"{prefix}DB_PORT", "5432")	×
43	db_name = os.environ.get(f"{prefix}DB_NAME", "renku")	×
44	pg_password = os.environ.get(f"{prefix}DB_PASSWORD")	×
45	if pg_password is None:	×
46	raise errors.ConfigurationError(	×
47	message="Please provide a database password in the 'DB_PASSWORD' environment variable."
48	)
49	async_sqlalchemy_url = f"postgresql+asyncpg://{pg_user}:{pg_password}@{pg_host}:{pg_port}/{db_name}"	×
50	# NOTE: the pool here is not used to serve HTTP requests, it is only used in background jobs.
51	# Therefore, we want to consume very few connections and we can wait for an available connection
52	# much longer than the default 30 seconds. In our tests syncing 15 users times out with the default.
53	engine = create_async_engine(async_sqlalchemy_url, pool_size=4, max_overflow=0, pool_timeout=600)	×
54	session_maker: Callable[..., AsyncSession] = sessionmaker(engine, class_=AsyncSession, expire_on_commit=False) # type: ignore[call-overload]	×
55	redis = RedisConfig.from_env(prefix)	×
56	message_queue = RedisQueue(redis)	×
57
58	authz_config = AuthzConfig.from_env()	×
59	event_repo = EventRepository(session_maker=session_maker, message_queue=message_queue)	×
60	group_repo = GroupRepository(	×
61	session_maker,
62	event_repo=event_repo,
63	group_authz=Authz(authz_config),
64	message_queue=message_queue,
65	)
66	project_repo = ProjectRepository(	×
67	session_maker=session_maker,
68	message_queue=message_queue,
69	event_repo=event_repo,
70	group_repo=group_repo,
71	authz=Authz(authz_config),
72	)
NEW 73	data_connector_repo = DataConnectorRepository(	×
74	session_maker=session_maker,
75	authz=Authz(authz_config),
76	)
NEW 77	data_connector_project_link_repo = DataConnectorProjectLinkRepository(	×
78	session_maker=session_maker,
79	authz=Authz(authz_config),
80	)
NEW 81	data_connector_migration_tool = DataConnectorMigrationTool(	×
82	session_maker=session_maker,
83	data_connector_repo=data_connector_repo,
84	data_connector_project_link_repo=data_connector_project_link_repo,
85	project_repo=project_repo,
86	authz=Authz(authz_config),
87	)
UNCOV 88	user_repo = UserRepo(	×
89	session_maker=session_maker,
90	message_queue=message_queue,
91	event_repo=event_repo,
92	group_repo=group_repo,
93	encryption_key=None,
94	authz=Authz(authz_config),
95	)
96	syncer = UsersSync(	×
97	session_maker,
98	message_queue=message_queue,
99	event_repo=event_repo,
100	group_repo=group_repo,
101	user_repo=user_repo,
102	authz=Authz(authz_config),
103	)
104	keycloak_url = os.environ[f"{prefix}KEYCLOAK_URL"]	×
105	client_id = os.environ[f"{prefix}KEYCLOAK_CLIENT_ID"]	×
106	client_secret = os.environ[f"{prefix}KEYCLOAK_CLIENT_SECRET"]	×
107	realm = os.environ.get(f"{prefix}KEYCLOAK_REALM", "Renku")	×
108	kc_api = KeycloakAPI(keycloak_url=keycloak_url, client_id=client_id, client_secret=client_secret, realm=realm)	×
NEW 109	return cls(	×
110	syncer,
111	kc_api,
112	authz_config,
113	group_repo,
114	event_repo,
115	project_repo,
116	data_connector_migration_tool,
117	session_maker,
118	)

SwissDataScienceCenter / renku-data-services / 11288123511

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous