diff --git a/bases/renku_data_services/data_api/app.py b/bases/renku_data_services/data_api/app.py index 5d3a23cf1..c91a67f22 100644 --- a/bases/renku_data_services/data_api/app.py +++ b/bases/renku_data_services/data_api/app.py @@ -144,6 +144,7 @@ def register_all_handlers(app: Sanic, dm: DependencyManager) -> Sanic: session_repo=dm.session_repo, session_secret_repo=dm.project_session_secret_repo, metrics=dm.metrics, + project_storage_k8s=dm.project_storage_k8s, ) project_session_secrets = ProjectSessionSecretBP( name="project_session_secrets", @@ -229,6 +230,7 @@ def register_all_handlers(app: Sanic, dm: DependencyManager) -> Sanic: internal_token_mint=dm.internal_token_mint, resource_usage_service=dm.resource_usage_service, resource_requests_repo=dm.resource_requests_repo, + authz=dm.authz, ) platform_config = PlatformConfigBP( name="platform_config", @@ -278,6 +280,7 @@ def register_all_handlers(app: Sanic, dm: DependencyManager) -> Sanic: data_service_base_url=dm.config.nb_config.data_service_url, k8s_client=dm.k8s_client, deposit_config=dm.config.deposit_config, + project_storage_k8s=dm.project_storage_k8s, ) notifications = NotificationsBP( name="notifications", diff --git a/bases/renku_data_services/data_api/config.py b/bases/renku_data_services/data_api/config.py index 045176de5..1994faeaf 100644 --- a/bases/renku_data_services/data_api/config.py +++ b/bases/renku_data_services/data_api/config.py @@ -14,7 +14,7 @@ ) from renku_data_services.app_config.logging import Config as LoggingConfig from renku_data_services.authz.config import AuthzConfig -from renku_data_services.data_connectors.config import DepositConfig +from renku_data_services.data_connectors.config import DepositConfig, ProjectStorageConfig from renku_data_services.db_config.config import DBConfig from renku_data_services.notebooks.config import NotebooksConfig from renku_data_services.secrets.config import PublicSecretsConfig @@ -48,6 +48,7 @@ class Config: version: str alertmanager_webhook_role: str deposit_config: DepositConfig + project_storage_config: ProjectStorageConfig @classmethod def from_env(cls, db: DBConfig | None = None) -> Self: @@ -95,4 +96,5 @@ def from_env(cls, db: DBConfig | None = None) -> Self: log_cfg=LoggingConfig.from_env(), alertmanager_webhook_role=os.environ.get("ALERTMANAGER_WEBHOOK_ROLE", "alertmanager-webhook"), deposit_config=DepositConfig.from_env(nb_config.sessions.renku_url), + project_storage_config=ProjectStorageConfig.from_env(), ) diff --git a/bases/renku_data_services/data_api/dependencies.py b/bases/renku_data_services/data_api/dependencies.py index 6c4200a39..aeda0d999 100644 --- a/bases/renku_data_services/data_api/dependencies.py +++ b/bases/renku_data_services/data_api/dependencies.py @@ -40,6 +40,7 @@ ) from renku_data_services.data_connectors.deposits.envidat import EnvidatClient from renku_data_services.data_connectors.deposits.zenodo import ZenodoAPIClient +from renku_data_services.data_connectors.project_storage_k8s import ProjectStorageK8s from renku_data_services.git.gitlab import DummyGitlabAPI, EmptyGitlabAPI, GitlabAPI from renku_data_services.k8s.client_interfaces import K8sClient from renku_data_services.k8s.clients import ( @@ -175,6 +176,7 @@ class DependencyManager: secret_client: K8sSecretClient internal_token_mint: RenkuSelfTokenMint internal_scope_verifier: ScopeVerifier + project_storage_k8s: ProjectStorageK8s spec: dict[str, Any] = field(init=False, repr=False, default_factory=dict) app_name: str = "renku_data_services" @@ -364,6 +366,7 @@ def from_env(cls) -> DependencyManager: resource_requests_repo=resource_requests_repo, member_repo=member_repo, ) + project_storage_k8s = ProjectStorageK8s(config.nb_config.k8s_v2_client) reprovisioning_repo = ReprovisioningRepository(session_maker=config.db.async_session_maker) git_repositories_repo = GitRepositoriesRepository( @@ -420,6 +423,7 @@ def from_env(cls) -> DependencyManager: project_repo=project_repo, group_repo=group_repo, search_updates_repo=search_updates_repo, + project_storage_config=config.project_storage_config, ) data_connector_secret_repo = DataConnectorSecretRepository( session_maker=config.db.async_session_maker, @@ -513,4 +517,5 @@ def from_env(cls) -> DependencyManager: secret_client=secret_client, internal_token_mint=internal_token_mint, internal_scope_verifier=internal_scope_verifier, + project_storage_k8s=project_storage_k8s, ) diff --git a/components/renku_data_services/base_models/bytesize.py b/components/renku_data_services/base_models/bytesize.py new file mode 100644 index 000000000..d55ca6c5c --- /dev/null +++ b/components/renku_data_services/base_models/bytesize.py @@ -0,0 +1,108 @@ +"""Byte size model with unit conversions.""" + +from __future__ import annotations + +from dataclasses import dataclass + + +@dataclass(frozen=True, order=True) +class ByteSize: + """Represents a size in bytes, with convenience conversions and formatting.""" + + value: int + + # Binary (1024-based) unit thresholds + KIBI = 1024 + MEBI = 1024**2 + GIBI = 1024**3 + TEBI = 1024**4 + + def __post_init__(self) -> None: + if self.value < 0: + raise ValueError(f"ByteSize cannot be negative: {self.value}") + if not isinstance(self.value, int): + raise TypeError(f"ByteSize value must be int, got {type(self.value).__name__}") + + def to_bytes(self) -> int: + """Return the size in bytes.""" + return self.value + + def to_kibi(self) -> float: + """Return the size in kibibytes (KiB).""" + return self.value / self.KIBI + + def to_mibi(self) -> float: + """Return the size in mebibytes (MiB).""" + return self.value / self.MEBI + + def to_gibi(self) -> float: + """Return the size in gibibytes (GiB).""" + return self.value / self.GIBI + + def to_tebi(self) -> float: + """Return the size in tebibytes (TiB).""" + return self.value / self.TEBI + + def to_human(self) -> str: + """Return a human-readable string with the appropriate binary unit.""" + if self.value < self.KIBI: + return f"{self.value}B" + elif self.value < self.MEBI: + return f"{self.to_kibi():.2f}KiB" + elif self.value < self.GIBI: + return f"{self.to_mibi():.2f}MiB" + elif self.value < self.TEBI: + return f"{self.to_gibi():.2f}GiB" + else: + return f"{self.to_tebi():.2f}TiB" + + def __str__(self) -> str: + return self.to_human() + + def __repr__(self) -> str: + return f"ByteSize({self.value}B)" + + def __add__(self, other: ByteSize) -> ByteSize: + return ByteSize(self.value + other.value) + + def __sub__(self, other: ByteSize) -> ByteSize: + result = self.value - other.value + if result < 0: + raise ValueError("Subtraction would result in negative ByteSize") + return ByteSize(result) + + def __radd__(self, other: int) -> ByteSize: + # allows sum([ByteSize(1), ByteSize(2)]) to work, since sum() starts with 0 + if other == 0: + return self + return NotImplemented + + @classmethod + def from_bytes(cls, bs: int) -> ByteSize: + """Create a ByteSize from a byte count.""" + return ByteSize(value=bs) + + @classmethod + def from_kibi(cls, kb: float) -> ByteSize: + """Create a ByteSize from a kibibyte value.""" + return ByteSize(value=int(kb * cls.KIBI)) + + @classmethod + def from_mibi(cls, mb: float) -> ByteSize: + """Create a ByteSize from a mebibyte value.""" + return ByteSize(value=int(mb * cls.MEBI)) + + @classmethod + def from_gibi(cls, gb: float) -> ByteSize: + """Create a ByteSize from a gibibyte value.""" + return ByteSize(value=int(gb * cls.GIBI)) + + @classmethod + def from_tebi(cls, tib: float) -> ByteSize: + """Create a ByteSize from a tebibyte value.""" + return cls(value=round(tib * cls.TEBI)) + + @classmethod + def zero(cls) -> ByteSize: + """Create a byte size with value 0.""" + return ByteSize(0) diff --git a/components/renku_data_services/base_models/core.py b/components/renku_data_services/base_models/core.py index 9995871d8..c31348fff 100644 --- a/components/renku_data_services/base_models/core.py +++ b/components/renku_data_services/base_models/core.py @@ -316,6 +316,12 @@ def from_strings(cls, *slugs: str) -> Self: raise errors.ValidationError(message=f"Two slug strings are needed to create a project path, got {slugs}.") return cls(NamespaceSlug(slugs[0]), ProjectSlug(slugs[1])) + @classmethod + def parse(cls, slug: str) -> Self: + """Parses a single string into a ProjectPath.""" + namespace_split = slug.split("/") + return cls.from_strings(*namespace_split) + @dataclass(frozen=True, eq=True, repr=False) class DataConnectorPath(__NamespaceCommonMixin): diff --git a/components/renku_data_services/data_connectors/api.spec.yaml b/components/renku_data_services/data_connectors/api.spec.yaml index 011afa453..215bfce7a 100644 --- a/components/renku_data_services/data_connectors/api.spec.yaml +++ b/components/renku_data_services/data_connectors/api.spec.yaml @@ -70,6 +70,226 @@ paths: $ref: "#/components/responses/Error" tags: - data_connectors + /data_connectors/storage: + post: + summary: Create a new project storage + requestBody: + required: true + content: + application/json: + schema: + $ref: "#/components/schemas/ProjectStoragePost" + responses: + "201": + description: The data connector was created + content: + application/json: + schema: + $ref: "#/components/schemas/ProjectStorage" + default: + $ref: "#/components/responses/Error" + tags: + - data_connectors + /data_connectors/storage/config: + get: + summary: Get the current configuration for project storage + responses: + "200": + description: The configuration data + content: + application/json: + schema: + $ref: "#/components/schemas/ProjectStorageConfig" + default: + $ref: "#/components/responses/Error" + tags: + - data_connectors + /data_connectors/storage/allow: + get: + summary: List all projects in the storage allow list + parameters: + - in: query + description: query parameters + name: params + style: form + explode: true + schema: + $ref: "#/components/schemas/ProjectStorageAllowListQuery" + responses: + "200": + description: List of storage allow entries + content: + application/json: + schema: + $ref: "#/components/schemas/ProjectStorageAllowList" + headers: + page: + description: The index of the current page (starting at 1). + required: true + schema: + type: integer + per-page: + description: The number of items per page. + required: true + schema: + type: integer + total: + description: The total number of items. + required: true + schema: + type: integer + total-pages: + description: The total number of pages. + required: true + schema: + type: integer + default: + $ref: "#/components/responses/Error" + tags: + - data_connectors + post: + summary: Add a project to the storage allow list + requestBody: + required: true + content: + application/json: + schema: + $ref: "#/components/schemas/ProjectStorageAllowPost" + responses: + "201": + description: The project was added to the allow list + content: + application/json: + schema: + $ref: "#/components/schemas/ProjectStorageAllowPost" + default: + $ref: "#/components/responses/Error" + tags: + - data_connectors + /data_connectors/storage/allow/{project_id}: + parameters: + - in: path + name: project_id + required: true + schema: + $ref: "#/components/schemas/Ulid" + get: + summary: Get the storage allow entry for a project + responses: + "200": + description: The project storage allow entry + content: + application/json: + schema: + $ref: "#/components/schemas/ProjectStorageAllow" + "404": + description: The project is not in the storage allow list + content: + application/json: + schema: + $ref: "#/components/schemas/ErrorResponse" + default: + $ref: "#/components/responses/Error" + tags: + - data_connectors + patch: + summary: Change the maximum size for a project storage + parameters: + - $ref: "#/components/parameters/If-Match" + requestBody: + required: true + content: + application/json: + schema: + $ref: "#/components/schemas/ProjectStorageAllowPatch" + responses: + "200": + description: The patched project storage allow entry + content: + application/json: + schema: + $ref: "#/components/schemas/ProjectStorageAllow" + "404": + description: The project storage allow entry doesn't exist + content: + application/json: + schema: + $ref: "#/components/schemas/ErrorResponse" + default: + $ref: "#/components/responses/Error" + tags: + - data_connectors + delete: + summary: Remove a project from the storage allow list + responses: + "204": + description: The project was removed from the allow list + default: + $ref: "#/components/responses/Error" + tags: + - data_connectors + /data_connectors/storage/{storage_id}: + parameters: + - in: path + name: storage_id + required: true + schema: + $ref: "#/components/schemas/Ulid" + get: + summary: Get a project storage for a project id. + responses: + "200": + description: The project storage information + content: + application/json: + schema: + $ref: "#/components/schemas/ProjectStorage" + "404": + description: The project storage does not exist + content: + application/json: + schema: + $ref: "#/components/schemas/ErrorResponse" + default: + $ref: "#/components/responses/Error" + tags: + - data_connectors + patch: + summary: Change the size or mount path of a project storage + parameters: + - $ref: "#/components/parameters/If-Match" + requestBody: + required: true + content: + application/json: + schema: + $ref: "#/components/schemas/ProjectStoragePatch" + responses: + "200": + description: The patched project storage entry + content: + application/json: + schema: + $ref: "#/components/schemas/ProjectStorage" + "404": + description: The project storage does not exist + content: + application/json: + schema: + $ref: "#/components/schemas/ErrorResponse" + default: + $ref: "#/components/responses/Error" + tags: + - data_connectors + delete: + summary: Delete a specific project storage + responses: + "204": + description: The project storage was deleted or did not exist in the first place + default: + $ref: "#/components/responses/Error" + tags: + - data_connectors /data_connectors/global: post: summary: Create a new data connector @@ -482,6 +702,26 @@ paths: $ref: "#/components/responses/Error" tags: - data_connectors + /projects/{project_id}/storage: + parameters: + - in: path + name: project_id + required: true + schema: + $ref: "#/components/schemas/Ulid" + get: + summary: Get the project storage for a given project + responses: + "200": + description: The list of project storages (currently either one or empty). + content: + application/json: + schema: + $ref: "#/components/schemas/ProjectStorageList" + default: + $ref: "#/components/responses/Error" + tags: + - projects /projects/{project_id}/data_connector_links: parameters: - in: path @@ -656,6 +896,14 @@ paths: - deposit components: schemas: + ProjectSlug: + type: string + description: The slug used to identify a project + minLength: 3 + example: "user/my-project" + # limitations based on allowed characters in project slugs from Gitlab from here: + # https://docs.gitlab.com/ee/user/reserved_names.html + pattern: "^[a-zA-Z0-9]+([_.\\-/][a-zA-Z0-9]+)*[_.\\-/]?[a-zA-Z0-9]$" DataConnectorsList: description: A list of data connectors type: array @@ -744,6 +992,144 @@ components: - $ref: "#/components/schemas/CloudStorageUrlV2" required: - storage + ProjectStoragePost: + description: | + A special data connector for associating a shared read+write + volume. There can be exactly one such storage per project. + additionalProperties: false + properties: + namespace: + $ref: "#/components/schemas/ProjectSlug" + size: + type: integer + format: int32 + mount_path: + type: string + required: + - namespace + - size + - mount_path + ProjectStorage: + description: | + A stored shared storage configuration for a project. + additionalProperties: false + properties: + id: + $ref: "#/components/schemas/Ulid" + project_id: + $ref: "#/components/schemas/Ulid" + size: + type: integer + format: int32 + mount_path: + type: string + creation_date: + $ref: "#/components/schemas/CreationDate" + created_by: + $ref: "#/components/schemas/UserId" + updated_at: + $ref: "#/components/schemas/CreationDate" + etag: + $ref: "#/components/schemas/ETag" + required: + - id + - project_id + - size + - mount_path + - creation_date + - created_by + - updated_at + - etag + ProjectStoragePatch: + description: | + Data that can be updated on a project storage entry + type: object + additionalProperties: false + properties: + size: + type: integer + format: int64 + description: The maximum size in GB + mount_path: + type: string + description: The mount path for the storage + ProjectStorageAllowPatch: + description: | + Data that can be updated on a project storage allow entry + type: object + additionalProperties: false + properties: + max_size: + type: integer + format: int64 + description: The maximum size in GB + ProjectStorageAllowPost: + description: | + A project to be added to the storage allow list. + additionalProperties: false + properties: + project_id: + $ref: "#/components/schemas/Ulid" + max_size: + type: integer + format: int64 + description: Maximum size in GB + required: + - project_id + - max_size + ProjectStorageAllow: + description: | + A project in the storage allow list. + additionalProperties: false + properties: + project_id: + $ref: "#/components/schemas/Ulid" + name: + type: string + namespace: + type: string + max_size: + type: integer + format: int64 + description: Maximum size in GB + etag: + $ref: "#/components/schemas/ETag" + required: + - project_id + - max_size + - name + - namespace + - etag + ProjectStorageAllowList: + description: | + A list of project storage allow entries. + type: array + items: + $ref: "#/components/schemas/ProjectStorageAllow" + ProjectStorageAllowListQuery: + description: Query params for listing storage allow entries + allOf: + - $ref: "#/components/schemas/PaginationRequest" + - properties: + project_name: + description: Filter by project name (partial match). + type: string + default: "" + ProjectStorageConfig: + description: | + The current configuration for project storage as defined by admins. + type: object + additionalProperties: false + properties: + enabled: + type: boolean + max_size: + type: integer + format: int64 + description: The maximum size in GB + required: + - enabled + - max_size DataConnectorPatch: description: | Patch of a data connector @@ -853,6 +1239,11 @@ components: type: array items: $ref: "#/components/schemas/DataConnectorToProjectLink" + ProjectStorageList: + description: A list of project storages. + type: array + items: + $ref: "#/components/schemas/ProjectStorage" DataConnectorToProjectLink: description: A link from a data connector to a project in Renku 2.0 type: object diff --git a/components/renku_data_services/data_connectors/apispec.py b/components/renku_data_services/data_connectors/apispec.py index 8d03c2f6f..54bf10623 100644 --- a/components/renku_data_services/data_connectors/apispec.py +++ b/components/renku_data_services/data_connectors/apispec.py @@ -1,6 +1,6 @@ # generated by datamodel-codegen: # filename: api.spec.yaml -# timestamp: 2026-07-13T12:25:32+00:00 +# timestamp: 2026-07-30T10:50:42+00:00 from __future__ import annotations @@ -11,6 +11,44 @@ from renku_data_services.data_connectors.apispec_base import BaseAPISpec +class ProjectStoragePost(BaseAPISpec): + model_config = ConfigDict( + extra="forbid", + ) + namespace: str = Field( + ..., + description="The slug used to identify a project", + examples=["user/my-project"], + min_length=3, + pattern="^[a-zA-Z0-9]+([_.\\-/][a-zA-Z0-9]+)*[_.\\-/]?[a-zA-Z0-9]$", + ) + size: int + mount_path: str + + +class ProjectStoragePatch(BaseAPISpec): + model_config = ConfigDict( + extra="forbid", + ) + size: int | None = Field(None, description="The maximum size in GB") + mount_path: str | None = Field(None, description="The mount path for the storage") + + +class ProjectStorageAllowPatch(BaseAPISpec): + model_config = ConfigDict( + extra="forbid", + ) + max_size: int | None = Field(None, description="The maximum size in GB") + + +class ProjectStorageConfig(BaseAPISpec): + model_config = ConfigDict( + extra="forbid", + ) + enabled: bool + max_size: int = Field(..., description="The maximum size in GB") + + class Example(BaseAPISpec): value: str = Field(..., description="a potential value for the option (think enum)") help: str = Field(..., description="help text for the value") @@ -253,6 +291,90 @@ class DepositsGetParametersQuery(BaseAPISpec): params: PaginationRequest | None = None +class ProjectStorage(BaseAPISpec): + model_config = ConfigDict( + extra="forbid", + ) + id: str = Field( + ..., + description="ULID identifier", + max_length=26, + min_length=26, + pattern="^[0-7][0-9A-HJKMNP-TV-Z]{25}$", + ) + project_id: str = Field( + ..., + description="ULID identifier", + max_length=26, + min_length=26, + pattern="^[0-7][0-9A-HJKMNP-TV-Z]{25}$", + ) + size: int + mount_path: str + creation_date: AwareDatetime = Field( + ..., + description="The date and time the resource was created (in UTC and ISO-8601 format)", + examples=["2023-11-01T17:32:28Z"], + ) + created_by: str = Field( + ..., + description="Keycloak user ID", + examples=["f74a228b-1790-4276-af5f-25c2424e9b0c"], + pattern="^[A-Za-z0-9]{1}[A-Za-z0-9-]+$", + ) + updated_at: AwareDatetime = Field( + ..., + description="The date and time the resource was created (in UTC and ISO-8601 format)", + examples=["2023-11-01T17:32:28Z"], + ) + etag: str = Field( + ..., description="Entity Tag", examples=["9EE498F9D565D0C41E511377425F32F3"] + ) + + +class ProjectStorageAllowPost(BaseAPISpec): + model_config = ConfigDict( + extra="forbid", + ) + project_id: str = Field( + ..., + description="ULID identifier", + max_length=26, + min_length=26, + pattern="^[0-7][0-9A-HJKMNP-TV-Z]{25}$", + ) + max_size: int = Field(..., description="Maximum size in GB") + + +class ProjectStorageAllow(BaseAPISpec): + model_config = ConfigDict( + extra="forbid", + ) + project_id: str = Field( + ..., + description="ULID identifier", + max_length=26, + min_length=26, + pattern="^[0-7][0-9A-HJKMNP-TV-Z]{25}$", + ) + name: str + namespace: str + max_size: int = Field(..., description="Maximum size in GB") + etag: str = Field( + ..., description="Entity Tag", examples=["9EE498F9D565D0C41E511377425F32F3"] + ) + + +class ProjectStorageAllowList(RootModel[list[ProjectStorageAllow]]): + root: list[ProjectStorageAllow] = Field( + ..., description="A list of project storage allow entries.\n" + ) + + +class ProjectStorageAllowListQuery(PaginationRequest): + project_name: str = Field("", description="Filter by project name (partial match).") + + class CloudStorageCore(BaseAPISpec): model_config = ConfigDict( extra="forbid", @@ -338,6 +460,10 @@ class CloudStorageUrlV2(BaseAPISpec): ) +class ProjectStorageList(RootModel[list[ProjectStorage]]): + root: list[ProjectStorage] = Field(..., description="A list of project storages.") + + class DataConnectorToProjectLink(BaseAPISpec): model_config = ConfigDict( extra="forbid", @@ -437,6 +563,10 @@ class DataConnectorsGetParametersQuery(BaseAPISpec): params: DataConnectorsGetQuery | None = None +class DataConnectorsStorageAllowGetParametersQuery(BaseAPISpec): + params: ProjectStorageAllowListQuery | None = None + + class DataConnectorLinksGetParametersQuery(BaseAPISpec): params: DataConnectorLinksGetQuery | None = None diff --git a/components/renku_data_services/data_connectors/blueprints.py b/components/renku_data_services/data_connectors/blueprints.py index a5107f818..2cb2d5077 100644 --- a/components/renku_data_services/data_connectors/blueprints.py +++ b/components/renku_data_services/data_connectors/blueprints.py @@ -11,6 +11,7 @@ from renku_data_services import base_models, errors from renku_data_services.base_api.auth import ( authenticate, + only_admins, only_authenticated, ) from renku_data_services.base_api.blueprint import BlueprintFactoryResponse, CustomBlueprint @@ -44,7 +45,11 @@ validate_deposit, validate_deposit_patch, validate_deposit_status_change, + validate_project_storage_allow_patch, + validate_project_storage_allow_post, + validate_project_storage_patch, validate_unsaved_data_connector, + validate_unsaved_project_storage, ) from renku_data_services.data_connectors.db import ( DOI, @@ -53,6 +58,7 @@ ) from renku_data_services.data_connectors.deposits.envidat import EnvidatClient from renku_data_services.data_connectors.deposits.zenodo import ZenodoAPIClient +from renku_data_services.data_connectors.project_storage_k8s import ProjectStorageK8s from renku_data_services.k8s.client_interfaces import K8sClient, SecretClient from renku_data_services.k8s.clients import DepositUploadJobClient from renku_data_services.notebooks.data_sources import DataSourceRepository @@ -77,6 +83,7 @@ class DataConnectorsBP(CustomBlueprint): data_service_base_url: str k8s_client: K8sClient deposit_config: DepositConfig + project_storage_k8s: ProjectStorageK8s def get_all(self) -> BlueprintFactoryResponse: """List data connectors.""" @@ -140,6 +147,183 @@ async def _post( return "/data_connectors", ["POST"], _post + def get_one_storage(self) -> BlueprintFactoryResponse: + """Get a specific project storage connector.""" + + @authenticate(self.authenticator) + @extract_if_none_match + async def _get_one(_: Request, user: base_models.APIUser, storage_id: ULID, etag: str | None) -> HTTPResponse: + project_storage = await self.data_connector_repo.get_project_storage(user=user, storage_id=storage_id) + if project_storage is None: + raise errors.MissingResourceError(message=f"No project storage found for storage: {storage_id}") + + if project_storage.etag == etag: + return HTTPResponse(status=304) + + headers = {"ETag": project_storage.etag} + return validated_json( + apispec.ProjectStorage, + self._dump_project_storage(project_storage), + headers=headers, + ) + + return "/data_connectors/storage/", ["GET"], _get_one + + def post_storage(self) -> BlueprintFactoryResponse: + """Create a new shared project storage.""" + + @authenticate(self.authenticator) + @only_authenticated + @validate(json=apispec.ProjectStoragePost) + async def _post_storage( + _: Request, user: base_models.APIUser, body: apispec.ProjectStoragePost + ) -> JSONResponse: + dc = validate_unsaved_project_storage(body) + result = await self.data_connector_repo.insert_project_storage(user, dc) + headers = {"ETag": result.etag} + return validated_json( + apispec.ProjectStorage, self._dump_project_storage(result), headers=headers, status=201 + ) + + return "/data_connectors/storage", ["POST"], _post_storage + + def patch_storage(self) -> BlueprintFactoryResponse: + """Partially update a project storage entry.""" + + @authenticate(self.authenticator) + @only_authenticated + @if_match_required + @validate(json=apispec.ProjectStoragePatch) + async def _patch_storage( + _: Request, + user: base_models.APIUser, + storage_id: ULID, + body: apispec.ProjectStoragePatch, + etag: str, + ) -> JSONResponse: + existing_storage = await self.data_connector_repo.get_project_storage(user=user, storage_id=storage_id) + if existing_storage is None: + raise errors.MissingResourceError(message=f"No project storage found for storage: {storage_id}") + + storage_patch = validate_project_storage_patch(existing_storage, body) + updated_storage = await self.data_connector_repo.update_project_storage( + user=user, storage_id=storage_id, patch=storage_patch, etag=etag + ) + headers = {"ETag": updated_storage.etag} + return validated_json( + apispec.ProjectStorage, + self._dump_project_storage(updated_storage), + headers=headers, + ) + + return "/data_connectors/storage/", ["PATCH"], _patch_storage + + def get_all_storage_allows(self) -> BlueprintFactoryResponse: + """List all projects in the storage allow list.""" + + @authenticate(self.authenticator) + @only_admins + @validate_query(query=apispec.ProjectStorageAllowListQuery) + @paginate + async def _get_all_storage_allows( + _: Request, + user: base_models.APIUser, + pagination: PaginationRequest, + query: apispec.ProjectStorageAllowListQuery, + ) -> tuple[list[dict[str, Any]], int]: + project_name = query.project_name if query.project_name else None + allows, total = await self.data_connector_repo.get_project_storage_allows( + user, pagination, project_name=project_name + ) + return [ + validate_and_dump( + apispec.ProjectStorageAllow, + self._dump_project_storage_allow_detail(a), + ) + for a in allows + ], total + + return "/data_connectors/storage/allow", ["GET"], _get_all_storage_allows + + def post_storage_allow(self) -> BlueprintFactoryResponse: + """Add a project to the storage allow list.""" + + @authenticate(self.authenticator) + @only_admins + @validate(json=apispec.ProjectStorageAllowPost) + async def _post_storage_allow( + _: Request, user: base_models.APIUser, body: apispec.ProjectStorageAllowPost + ) -> JSONResponse: + allow = validate_project_storage_allow_post(body) + inserted = await self.data_connector_repo.insert_project_storage_allow(user, allow) + return validated_json( + apispec.ProjectStorageAllowPost, + self._dump_project_storage_allow_post(inserted), + status=201, + ) + + return "/data_connectors/storage/allow", ["POST"], _post_storage_allow + + def patch_storage_allow(self) -> BlueprintFactoryResponse: + """Partially update a project storage allow entry.""" + + @authenticate(self.authenticator) + @only_admins + @if_match_required + @validate(json=apispec.ProjectStorageAllowPatch) + async def _patch( + _: Request, + user: base_models.APIUser, + project_id: ULID, + body: apispec.ProjectStorageAllowPatch, + etag: str, + ) -> JSONResponse: + existing_entry = await self.data_connector_repo.get_project_storage_allow(user=user, project_id=project_id) + if not existing_entry: + raise errors.MissingResourceError(message=f"No project storage allow entry for project {project_id}") + + pse_patch = validate_project_storage_allow_patch(existing_entry, body) + pse_update = await self.data_connector_repo.update_project_storage_allow( + user=user, project_id=project_id, patch=pse_patch, etag=etag + ) + + headers = {"ETag": pse_update.new.etag} + return validated_json( + apispec.ProjectStorageAllow, self._dump_project_storage_allow_detail(pse_update.new), headers=headers + ) + + return "/data_connectors/storage/allow/", ["PATCH"], _patch + + def get_storage_allow(self) -> BlueprintFactoryResponse: + """Get the storage allow entry for a project.""" + + @authenticate(self.authenticator) + @only_authenticated + async def _get_storage_allow(_: Request, user: base_models.APIUser, project_id: ULID) -> HTTPResponse: + allow = await self.data_connector_repo.get_project_storage_allow(user, project_id) + if allow is None: + raise errors.MissingResourceError(message=f"Project {project_id} is not in the storage allow list.") + + headers = {"ETag": allow.etag} + return validated_json( + apispec.ProjectStorageAllow, self._dump_project_storage_allow_detail(allow), headers=headers + ) + + return "/data_connectors/storage/allow/", ["GET"], _get_storage_allow + + def delete_storage_allow(self) -> BlueprintFactoryResponse: + """Remove a project from the storage allow list.""" + + @authenticate(self.authenticator) + @only_admins + async def _delete_storage_allow(_: Request, user: base_models.APIUser, project_id: ULID) -> HTTPResponse: + deleted = await self.data_connector_repo.delete_project_storage_allow(user, project_id) + if deleted: + await self.project_storage_k8s.delete_volume(deleted) + return HTTPResponse(status=204) + + return "/data_connectors/storage/allow/", ["DELETE"], _delete_storage_allow + def post_global(self) -> BlueprintFactoryResponse: """Create a new global data connector.""" @@ -396,6 +580,52 @@ async def _delete_project_link( _delete_project_link, ) + def get_storage_to_project(self) -> BlueprintFactoryResponse: + """List all project storage to a given project.""" + + @authenticate(self.authenticator) + async def _get_all_storage_to_project( + _: Request, + user: base_models.APIUser, + project_id: ULID, + ) -> JSONResponse: + project_storage = await self.data_connector_repo.get_storage_to(user=user, project_id=project_id) + result = [self._dump_project_storage(project_storage)] if project_storage else [] + return validated_json(apispec.ProjectStorageList, result) + + return "/projects//storage", ["GET"], _get_all_storage_to_project + + def get_storage_config(self) -> BlueprintFactoryResponse: + """Get the current config used for project storage.""" + + @authenticate(self.authenticator) + @only_admins + async def _get_project_config(_: Request, user: base_models.APIUser) -> JSONResponse: + storage_config = self.data_connector_repo.get_project_storage_config() + result = apispec.ProjectStorageConfig( + enabled=storage_config.enabled, max_size=int(storage_config.maximum_size.to_gibi()) + ) + return validated_json(apispec.ProjectStorageConfig, result) + + return "/data_connectors/storage/config", ["GET"], _get_project_config + + def delete_storage(self) -> BlueprintFactoryResponse: + """Delete a specific project storage.""" + + @authenticate(self.authenticator) + @only_authenticated + async def _delete_storage( + _: Request, + user: base_models.APIUser, + storage_id: ULID, + ) -> HTTPResponse: + deleted = await self.data_connector_repo.delete_project_storage(user=user, storage_id=storage_id) + if deleted: + await self.project_storage_k8s.delete_volume(deleted) + return HTTPResponse(status=204) + + return "/data_connectors/storage/", ["DELETE"], _delete_storage + def get_all_data_connectors_links_to_project(self) -> BlueprintFactoryResponse: """List all links from data connectors to a given project.""" @@ -593,6 +823,36 @@ def _dump_data_connector_secret(secret: models.DataConnectorSecret) -> dict[str, secret_id=str(secret.secret_id), ) + @staticmethod + def _dump_project_storage(ps: models.ProjectStorage) -> apispec.ProjectStorage: + return apispec.ProjectStorage( + id=str(ps.id), + project_id=str(ps.project_id), + size=int(ps.size.to_gibi()), + mount_path=ps.mount_path.as_posix(), + created_by=ps.created_by, + creation_date=ps.creation_date, + updated_at=ps.updated_at, + etag=ps.etag, + ) + + @staticmethod + def _dump_project_storage_allow_detail(ps: models.ProjectStorageAllowDetail) -> apispec.ProjectStorageAllow: + return apispec.ProjectStorageAllow( + project_id=str(ps.project_id), + max_size=int(ps.max_size.to_gibi()), + name=ps.name, + namespace=ps.namespace_path.serialize(), + etag=ps.etag, + ) + + @staticmethod + def _dump_project_storage_allow_post(ps: models.ProjectStorageAllow) -> apispec.ProjectStorageAllowPost: + return apispec.ProjectStorageAllowPost( + project_id=str(ps.project_id), + max_size=int(ps.max_size.to_gibi()), + ) + async def __get_zenodo_access_token(self, user: base_models.APIUser) -> str: provider = await self.connected_services_repo.get_provider_for_kind(user, ProviderKind.zenodo) if not provider: diff --git a/components/renku_data_services/data_connectors/config.py b/components/renku_data_services/data_connectors/config.py index f429405c0..ae1acac09 100644 --- a/components/renku_data_services/data_connectors/config.py +++ b/components/renku_data_services/data_connectors/config.py @@ -11,12 +11,39 @@ from kubernetes.client import ApiClient, V1Toleration from renku_data_services.app_config import logging +from renku_data_services.base_models.bytesize import ByteSize from renku_data_services.errors import errors from renku_data_services.k8s.constants import DEFAULT_K8S_CLUSTER, ClusterId logger = logging.getLogger(__name__) +@dataclass +class ProjectStorageConfig: + """The configuration for project storage.""" + + enabled: bool + storage_class: str + maximum_size: ByteSize + + @classmethod + def from_env(cls) -> ProjectStorageConfig: + """Create a configuration from environment variables.""" + + enabled = os.environ.get("PROJECT_STORAGE_ENABLED", "").lower() == "true" + storage_class = os.environ.get("PROJECT_STORAGE_STORAGE_CLASS") + maximum_size = os.environ.get("PROJECT_STORAGE_MAX_SIZE_GB") or "10" + maximum_size = ByteSize.from_gibi(int(maximum_size)) + + # TODO: set defaults for easier PR/CI deployments for now + enabled = True + storage_class = "azurefile" + if enabled and not storage_class: + raise errors.ConfigurationError(message="A storage_class is required for enabled project storage") + + return ProjectStorageConfig(enabled, storage_class=storage_class, maximum_size=maximum_size) + + @dataclass class DepositConfig: """The configuration for running data deposit uploads.""" diff --git a/components/renku_data_services/data_connectors/core.py b/components/renku_data_services/data_connectors/core.py index 05b675bdd..9d3ec0a6d 100644 --- a/components/renku_data_services/data_connectors/core.py +++ b/components/renku_data_services/data_connectors/core.py @@ -43,6 +43,7 @@ from renku_data_services import base_models, errors from renku_data_services.authz.models import Visibility +from renku_data_services.base_models.bytesize import ByteSize from renku_data_services.base_models.core import ( NamespacePath, ProjectPath, @@ -211,6 +212,84 @@ async def validate_unsaved_data_connector( ) +def _validate_mount_path(path: str | None) -> None: + # + invalid_prefixes = [ + "/", + "/bin", + "/sbin", + "/usr", + "/lib", + "/lib64", + "/boot", + "/etc", + "/proc", + "/sys", + "/dev", + "/run", + "/sys", + "/var", + "/tmp", # nosec B108 + "/home", + "/root", + ] + if not path or path == "": + raise errors.ValidationError(message="The mount path must not be empty") + + for prefix in invalid_prefixes: + if path == prefix or path.startswith(f"{prefix}/"): + raise errors.ValidationError(message=f"The mount path is invalid: '{path}'") + + +def validate_unsaved_project_storage(body: apispec.ProjectStoragePost) -> models.UnsavedProjectStorage: + """Validate the user input for a new project storage definition. + + The namespace must be a project namespace. The project must be + enabled for project storages and the user must be an owner. + """ + + _validate_mount_path(body.mount_path) + + namespace_path = ProjectPath.parse(body.namespace) + return models.UnsavedProjectStorage( + namespace_path=namespace_path, size=ByteSize.from_gibi(body.size), mount_path=PurePosixPath(body.mount_path) + ) + + +def validate_project_storage_patch( + existing: models.ProjectStorage, body: apispec.ProjectStoragePatch +) -> models.ProjectStoragePatch: + """Validate a patch of a project storage entry.""" + size = ByteSize.from_gibi(body.size) if body.size else None + if size and size < ByteSize.from_gibi(1): + raise errors.ValidationError(message="The size must be at least 1GB") + mount_path = PurePosixPath(body.mount_path) if body.mount_path else None + if mount_path: + _validate_mount_path(body.mount_path) + return models.ProjectStoragePatch(size=size, mount_path=mount_path) + + +def validate_project_storage_allow_post(body: apispec.ProjectStorageAllowPost) -> models.ProjectStorageAllow: + """Validate.""" + allow = models.ProjectStorageAllow( + project_id=ULID.from_str(body.project_id), max_size=ByteSize.from_gibi(body.max_size), updated_at=datetime.now() + ) + if allow.max_size < ByteSize.from_gibi(1): + raise errors.ValidationError(message=f"The maximum size must be at least 1GB, but {allow.max_size} was given.") + return allow + + +def validate_project_storage_allow_patch( + existing: models.ProjectStorageAllowDetail, body: apispec.ProjectStorageAllowPatch +) -> models.ProjectStorageAllowPatch: + """Validate a patch of a project storage allow entry.""" + + size = ByteSize.from_gibi(body.max_size) if body.max_size is not None else None + if size and size < ByteSize.from_gibi(1): + raise errors.ValidationError(message="The maximum size must be at least 1GB") + return models.ProjectStorageAllowPatch(max_size=size) + + async def prevalidate_unsaved_global_data_connector( body: apispec.GlobalDataConnectorPost, validator: RCloneValidator ) -> models.PrevalidatedGlobalDataConnector: diff --git a/components/renku_data_services/data_connectors/db.py b/components/renku_data_services/data_connectors/db.py index 2e2524453..400762748 100644 --- a/components/renku_data_services/data_connectors/db.py +++ b/components/renku_data_services/data_connectors/db.py @@ -8,7 +8,7 @@ from typing import TypeVar from cryptography.hazmat.primitives.asymmetric import rsa -from sqlalchemy import ColumnExpressionArgument, Select, delete, func, or_, select +from sqlalchemy import ColumnExpressionArgument, Select, and_, delete, exists, func, or_, select from sqlalchemy.ext.asyncio import AsyncSession from sqlalchemy.orm import joinedload from ulid import ULID @@ -29,6 +29,7 @@ ) from renku_data_services.data_connectors import apispec, models from renku_data_services.data_connectors import orm as schemas +from renku_data_services.data_connectors.config import ProjectStorageConfig from renku_data_services.data_connectors.core import validate_unsaved_global_data_connector from renku_data_services.data_connectors.doi.models import DOI from renku_data_services.k8s.constants import DEFAULT_K8S_CLUSTER @@ -44,7 +45,7 @@ from renku_data_services.secrets.models import SecretKind from renku_data_services.storage.rclone import RCloneValidator from renku_data_services.users.db import UserRepo -from renku_data_services.utils.core import with_db_transaction +from renku_data_services.utils.core import with_db_session, with_db_transaction class DataConnectorRepository: @@ -57,12 +58,14 @@ def __init__( project_repo: ProjectRepository, group_repo: GroupRepository, search_updates_repo: SearchUpdatesRepo, + project_storage_config: ProjectStorageConfig, ) -> None: self.session_maker = session_maker self.authz = authz self.project_repo = project_repo self.group_repo = group_repo self.search_updates_repo = search_updates_repo + self.project_storage_config = project_storage_config async def get_data_connectors( self, @@ -372,6 +375,390 @@ async def insert_namespaced_data_connector( raise errors.ProgrammingError(message=f"Expected to get a namespaced data connector ('{dc.id}')") return dc + async def get_project_storage(self, user: base_models.APIUser, storage_id: ULID) -> models.ProjectStorage | None: + """Get a project storage by its id.""" + + if user.id is None: + raise errors.UnauthorizedError(message="You do not have the required permissions for this operation.") + + async with self.session_maker() as session: + result_orm = await session.scalars( + select(schemas.ProjectStorageORM).where(schemas.ProjectStorageORM.id == storage_id) + ) + result_orm = result_orm.one_or_none() + if not result_orm: + return None + + result = result_orm.dump() + authorized = await self.authz.has_permission(user, ResourceType.project, result.project_id, Scope.READ) + if not authorized: + return None + + return result + + async def get_storage_to(self, user: base_models.APIUser, project_id: ULID) -> models.ProjectStorage | None: + """Get a project storage to a project if it exists and the feature is enabled.""" + + if not self.project_storage_config.enabled: + return None + else: + return await self._get_storage_to_project(user, project_id) + + async def _get_storage_to_project( + self, user: base_models.APIUser, project_id: ULID + ) -> models.ProjectStorage | None: + """Get a project storage to a project if it exists.""" + + if user.id is None: + raise errors.UnauthorizedError(message="You do not have the required permissions for this operation.") + + async with self.session_maker() as session: + result_orm = await session.scalars( + select(schemas.ProjectStorageORM).where(schemas.ProjectStorageORM.project_id == project_id) + ) + result_orm = result_orm.one_or_none() + if not result_orm: + return None + + result = result_orm.dump() + authorized = await self.authz.has_permission(user, ResourceType.project, result.project_id, Scope.READ) + if not authorized: + return None + + return result + + def get_project_storage_config(self) -> ProjectStorageConfig: + """Return the current config for project storage.""" + + return self.project_storage_config + + @with_db_transaction + async def insert_project_storage( + self, user: base_models.APIUser, input: models.UnsavedProjectStorage, *, session: AsyncSession | None = None + ) -> models.ProjectStorage: + """Insert a new project storage.""" + + # When the feature is disabled, we disallow insertion, but still allow managing existing data + if not self.project_storage_config.enabled: + raise errors.MissingResourceError(message="The project storage api is not enabled.") + + if not session: + raise errors.ProgrammingError(message="A database session is required.") + if user.id is None: + raise errors.UnauthorizedError(message="You do not have the required permissions for this operation.") + + # there is only one such storage possible for a project + project = await self.project_repo.get_project_by_namespace_slug( + user, input.namespace_path.first.value, input.namespace_path.second, with_documentation=False + ) + + authorized = await self.authz.has_permission(user, ResourceType.project, project.id, Scope.DELETE) + if not authorized: + raise errors.MissingResourceError( + message=f"Project with id '{project.id}' does not exist or you do not have access to it." + ) + + allowed = await session.execute( + select(schemas.ProjectStorageAllowORM).where(schemas.ProjectStorageAllowORM.project_id == project.id) + ) + allowed = allowed.scalar() + if not allowed: + raise errors.ForbiddenError(message=f"Project storage is not enabled for project {project.id}.") + + existing_storage = await session.execute( + select(exists().where(schemas.ProjectStorageORM.project_id == project.id)) + ) + existing_storage = existing_storage.scalar() + if existing_storage: + raise errors.ValidationError(message=f"There is already a project storage for project {project.id}") + + if input.size > allowed.max_size: + raise errors.ValidationError( + message=( + f"The project storage size ({input.size}) for project {project.id} " + f"exceeds the maximum size of {allowed.max_size}" + ) + ) + + new_storage = schemas.ProjectStorageORM( + project_id=project.id, + storage_class=self.project_storage_config.storage_class, + size_limit=input.size, + mount_path=input.mount_path, + created_by_id=user.id, + ) + session.add(new_storage) + await session.flush() + return new_storage.dump() + + @with_db_transaction + async def update_project_storage( + self, + user: base_models.APIUser, + storage_id: ULID, + patch: models.ProjectStoragePatch, + etag: str, + *, + session: AsyncSession | None = None, + ) -> models.ProjectStorage: + """Update some properties of a project storage entry.""" + if not session: + raise errors.ProgrammingError(message="A database session is required.") + + result = await session.scalars( + select(schemas.ProjectStorageORM).where(schemas.ProjectStorageORM.id == storage_id) + ) + storage_orm = result.one_or_none() + if storage_orm is None: + raise errors.MissingResourceError(message=f"Project storage with id '{storage_id}' does not exist.") + + # Check authorization - user must be "owner", meaning allowed to delete the project + authorized = await self.authz.has_permission(user, ResourceType.project, storage_orm.project_id, Scope.DELETE) + if not authorized: + raise errors.MissingResourceError( + message=f"Project storage with id '{storage_id}' does not exist or you do not have access to it." + ) + + current_storage = storage_orm.dump() + current_etag = current_storage.etag + if current_etag != etag: + raise errors.ConflictError(message=f"Current ETag is {current_etag}, not {etag}.") + + # Check if size would exceed the allowed maximum + new_size = patch.size if patch.size else current_storage.size + allowed = await session.execute( + select(schemas.ProjectStorageAllowORM).where( + schemas.ProjectStorageAllowORM.project_id == storage_orm.project_id + ) + ) + allowed = allowed.scalar() + if allowed and new_size > allowed.max_size: + raise errors.ValidationError( + message=( + f"The project storage size ({new_size}) for project {storage_orm.project_id} " + f"exceeds the maximum size of {allowed.max_size}" + ) + ) + + if patch.size is not None: + storage_orm.size_limit = patch.size + if patch.mount_path is not None: + storage_orm.mount_path = patch.mount_path + + await session.flush() + await session.refresh(storage_orm) + return storage_orm.dump() + + @with_db_transaction + async def insert_project_storage_allow( + self, user: base_models.APIUser, input: models.ProjectStorageAllow, *, session: AsyncSession | None = None + ) -> models.ProjectStorageAllow: + """Insert a new project storage allow entry.""" + if not session: + raise errors.ProgrammingError(message="A database session is required.") + if user.id is None or not user.is_admin: + raise errors.UnauthorizedError(message="You do not have the required permissions for this operation.") + + existing = await session.execute( + select(exists().where(schemas.ProjectStorageAllowORM.project_id == input.project_id)) + ) + if existing.scalar(): + raise errors.ValidationError(message=f"Project {input.project_id} is already in the allow list.") + + if input.max_size > self.project_storage_config.maximum_size: + raise errors.ValidationError( + message=( + f"The maximum size {input.max_size} exceeds the configured " + f"one of {self.project_storage_config.maximum_size}." + ) + ) + + new_allow = schemas.ProjectStorageAllowORM( + project_id=input.project_id, + max_size=input.max_size, + ) + session.add(new_allow) + await session.flush() + return new_allow.dump() + + @with_db_transaction + async def delete_project_storage( + self, user: base_models.APIUser, storage_id: ULID, *, session: AsyncSession | None = None + ) -> models.DeletedProjectStorage | None: + """Delete a specific project storage.""" + if not session: + raise errors.ProgrammingError(message="A database session is required.") + + result = await session.scalars( + select(schemas.ProjectStorageORM).where(schemas.ProjectStorageORM.id == storage_id) + ) + storage_orm = result.one_or_none() + if storage_orm is None: + return None + + authorized = await self.authz.has_permission(user, ResourceType.project, storage_orm.project_id, Scope.DELETE) + if not authorized: + raise errors.MissingResourceError( + message=f"Project storage with id '{storage_id}' does not exist or you do not have access to it." + ) + + await session.delete(storage_orm) + ps = storage_orm.dump() + return models.DeletedProjectStorage(project_id=ps.project_id) + + @with_db_session + async def get_project_storage_allow( + self, + user: base_models.APIUser, + project_id: ULID, + *, + session: AsyncSession | None = None, + ) -> models.ProjectStorageAllowDetail | None: + """Get the storage allow entry for a project if it exists.""" + if not session: + raise errors.ProgrammingError(message="A database session is required.") + + authorized = await self.authz.has_permission(user, ResourceType.project, project_id, Scope.READ) + if not authorized: + raise errors.MissingResourceError( + message=f"Project with id '{project_id}' does not exist or you do not have access to it." + ) + stmt = ( + select( + schemas.ProjectStorageAllowORM.project_id, + schemas.ProjectStorageAllowORM.max_size, + ProjectORM.name, + ns_schemas.NamespaceORM.slug.label("namespace_slug"), + ns_schemas.EntitySlugORM.slug.label("project_slug"), + schemas.ProjectStorageAllowORM.updated_at, + ) + .join(ProjectORM, ProjectORM.id == schemas.ProjectStorageAllowORM.project_id) + .join( + ns_schemas.EntitySlugORM, + and_( + ns_schemas.EntitySlugORM.project_id == schemas.ProjectStorageAllowORM.project_id, + ns_schemas.EntitySlugORM.data_connector_id.is_(None), + ), + ) + .join(ns_schemas.NamespaceORM, ns_schemas.NamespaceORM.id == ns_schemas.EntitySlugORM.namespace_id) + .where(schemas.ProjectStorageAllowORM.project_id == project_id) + ) + result = (await session.execute(stmt)).one_or_none() + if result: + return models.ProjectStorageAllowDetail.create(**result._mapping) + return None + + async def get_project_storage_allows( + self, user: base_models.APIUser, pagination: PaginationRequest, project_name: str | None = None + ) -> tuple[list[models.ProjectStorageAllowDetail], int]: + """Get all project storage allow entries, optionally filtered by project name.""" + if user.id is None or not user.is_admin: + raise errors.ForbiddenError(message="You do not have the required permissions for this operation.") + + async with self.session_maker() as session: + stmt = ( + select( + schemas.ProjectStorageAllowORM.project_id, + schemas.ProjectStorageAllowORM.max_size, + ProjectORM.name, + ns_schemas.NamespaceORM.slug.label("namespace_slug"), + ns_schemas.EntitySlugORM.slug.label("project_slug"), + schemas.ProjectStorageAllowORM.updated_at, + ) + .join(ProjectORM, ProjectORM.id == schemas.ProjectStorageAllowORM.project_id) + .join( + ns_schemas.EntitySlugORM, + and_( + ns_schemas.EntitySlugORM.project_id == schemas.ProjectStorageAllowORM.project_id, + ns_schemas.EntitySlugORM.data_connector_id.is_(None), + ), + ) + .join(ns_schemas.NamespaceORM, ns_schemas.NamespaceORM.id == ns_schemas.EntitySlugORM.namespace_id) + ) + + stmt_count = select(func.count()).select_from(schemas.ProjectStorageAllowORM) + if project_name: + stmt = stmt.where(ProjectORM.name.ilike(f"%{project_name}%")) + stmt_count = stmt_count.where(ProjectORM.name.ilike(f"%{project_name}%")) + stmt = ( + stmt.order_by(schemas.ProjectStorageAllowORM.project_id) + .limit(pagination.per_page) + .offset(pagination.offset) + ) + rows = await session.execute(stmt) + results = [models.ProjectStorageAllowDetail.create(**row._mapping) for row in rows] + total = await session.scalar(stmt_count) or 0 + return results, total + + @with_db_transaction + async def update_project_storage_allow( + self, + user: base_models.APIUser, + project_id: ULID, + patch: models.ProjectStorageAllowPatch, + etag: str, + *, + session: AsyncSession | None = None, + ) -> models.ProjectStorageAllowUpdate: + """Update some properties of a project storage allow entry.""" + if not session: + raise errors.ProgrammingError(message="A database session is required.") + + old = await self.get_project_storage_allow(user, project_id, session=session) + ps_orm = await session.scalars( + select(schemas.ProjectStorageAllowORM).where(schemas.ProjectStorageAllowORM.project_id == project_id) + ) + ps_orm = ps_orm.one_or_none() + if not old or not ps_orm or not isinstance(user, base_models.AuthenticatedAPIUser) or not user.is_admin: + raise errors.MissingResourceError( + message=( + f"Project storage allow entry for project '{project_id}' " + "does not exist or you do not have access to it." + ) + ) + + current_etag = old.etag + if current_etag != etag: + raise errors.ConflictError(message=f"Current ETag is {current_etag}, not {etag}.") + + if patch.max_size: + ps_orm.max_size = patch.max_size + + await session.flush() + await session.refresh(ps_orm) + + new = models.ProjectStorageAllowDetail( + project_id=project_id, + max_size=ps_orm.max_size, + name=old.name, + namespace_path=old.namespace_path, + updated_at=ps_orm.updated_at, + ) + return models.ProjectStorageAllowUpdate(old=old, new=new) + + @with_db_transaction + async def delete_project_storage_allow( + self, user: base_models.APIUser, project_id: ULID, *, session: AsyncSession | None = None + ) -> models.DeletedProjectStorage | None: + """Delete a project storage allow entry.""" + if not session: + raise errors.ProgrammingError(message="A database session is required.") + + if user.id is None or not user.is_admin: + raise errors.UnauthorizedError(message="You do not have the required permissions for this operation.") + + storage = await self._get_storage_to_project(user, project_id) + result = await session.scalars( + select(schemas.ProjectStorageAllowORM).where(schemas.ProjectStorageAllowORM.project_id == project_id) + ) + allow_orm = result.one_or_none() + if allow_orm: + await session.delete(allow_orm) + + if storage: + return models.DeletedProjectStorage(project_id=storage.project_id) + return None + @with_db_transaction async def insert_global_data_connector( self, diff --git a/components/renku_data_services/data_connectors/models.py b/components/renku_data_services/data_connectors/models.py index 23870fc2b..0f2aa2907 100644 --- a/components/renku_data_services/data_connectors/models.py +++ b/components/renku_data_services/data_connectors/models.py @@ -11,6 +11,7 @@ from ulid import ULID from renku_data_services.authz.models import Visibility +from renku_data_services.base_models.bytesize import ByteSize from renku_data_services.base_models.core import ( DataConnectorInProjectPath, DataConnectorPath, @@ -90,6 +91,112 @@ def path(self) -> DataConnectorPath | DataConnectorInProjectPath: return self.namespace / DataConnectorSlug(self.slug) +@dataclass(frozen=True, eq=True, kw_only=True) +class UnsavedProjectStorage: + """Project storage definition.""" + + namespace_path: ProjectPath + size: ByteSize + mount_path: PurePosixPath + + +@dataclass(frozen=True, eq=True, kw_only=True) +class ProjectStoragePatch: + """Model for changes requested on a project storage.""" + + size: ByteSize | None + mount_path: PurePosixPath | None + + +@dataclass(frozen=True, eq=True, kw_only=True) +class ProjectStorage: + """Stored project storage information.""" + + id: ULID + project_id: ULID + storage_class: str + size: ByteSize + mount_path: PurePosixPath + created_by: str + creation_date: datetime + updated_at: datetime + + @property + def etag(self) -> str: + """Entity tag value for this project storage object.""" + return compute_etag_from_fields( + self.updated_at, self.project_id, self.storage_class, self.size.to_bytes(), self.mount_path.as_posix() + ) + + +@dataclass(frozen=True, eq=True, kw_only=True) +class DeletedProjectStorage: + """A project storage that has been deleted.""" + + project_id: ULID + + +@dataclass(frozen=True, eq=True, kw_only=True) +class ProjectStorageAllow: + """Allowed project storage with max size.""" + + project_id: ULID + max_size: ByteSize + updated_at: datetime + + @property + def etag(self) -> str: + """Entity tag value for this project storage allow object.""" + return compute_etag_from_fields(self.updated_at, self.project_id, self.max_size.to_bytes()) + + +@dataclass(frozen=True, eq=True, kw_only=True) +class ProjectStorageAllowDetail: + """Allowed project storage with max size.""" + + project_id: ULID + max_size: ByteSize + name: str + namespace_path: ProjectPath + updated_at: datetime + + @classmethod + def create( + cls, + project_id: ULID, + max_size: ByteSize, + name: str, + namespace_slug: str, + project_slug: str, + updated_at: datetime, + ) -> ProjectStorageAllowDetail: + """Create an instance with the project path given as two strings.""" + np = ProjectPath.from_strings(namespace_slug, project_slug) + return ProjectStorageAllowDetail( + project_id=project_id, max_size=max_size, name=name, namespace_path=np, updated_at=updated_at + ) + + @property + def etag(self) -> str: + """Entity tag value for this project storage allow object.""" + return compute_etag_from_fields(self.updated_at, self.project_id, self.max_size.to_bytes()) + + +@dataclass(frozen=True, eq=True, kw_only=True) +class ProjectStorageAllowPatch: + """Model for changes requested on a project storage allow entry.""" + + max_size: ByteSize | None + + +@dataclass(frozen=True, eq=True, kw_only=True) +class ProjectStorageAllowUpdate: + """Return data when updating an allow entry.""" + + old: ProjectStorageAllowDetail + new: ProjectStorageAllowDetail + + @dataclass(frozen=True, eq=True, kw_only=True) class GlobalDataConnector(BaseDataConnector): """Global data connector model.""" diff --git a/components/renku_data_services/data_connectors/orm.py b/components/renku_data_services/data_connectors/orm.py index 5a8cd3196..94be7d007 100644 --- a/components/renku_data_services/data_connectors/orm.py +++ b/components/renku_data_services/data_connectors/orm.py @@ -13,6 +13,7 @@ from ulid import ULID from renku_data_services.authz import models as authz_models +from renku_data_services.base_models.bytesize import ByteSize from renku_data_services.base_orm.registry import COMMON_ORM_REGISTRY from renku_data_services.crc.orm import ClusterORM from renku_data_services.data_connectors import models @@ -22,7 +23,7 @@ from renku_data_services.project.orm import ProjectORM from renku_data_services.secrets.orm import SecretORM from renku_data_services.users.orm import UserORM -from renku_data_services.utils.sqlalchemy import ULIDType +from renku_data_services.utils.sqlalchemy import ByteSizeType, PurePosixPathType, ULIDType if TYPE_CHECKING: from renku_data_services.namespace.orm import EntitySlugOldORM, EntitySlugORM @@ -332,3 +333,89 @@ def dump(self) -> models.DepositJob: updated_at=self.updated_at, ), ) + + +class ProjectStorageAllowORM(BaseORM): + """ORM model for project storage allow list with size limits.""" + + __tablename__ = "project_storage_allow" + + project_id: Mapped[ULID] = mapped_column( + "project_id", + ForeignKey(ProjectORM.id, ondelete="CASCADE"), + primary_key=True, + unique=True, + index=True, + ) + """ID of the project.""" + + max_size: Mapped[ByteSize] = mapped_column("max_size", ByteSizeType()) + """Maximum allowed size in bytes.""" + + updated_at: Mapped[datetime] = mapped_column( + "updated_at", + DateTime(timezone=True), + default=None, + server_default=func.now(), + onupdate=func.now(), + nullable=False, + ) + + def dump(self) -> models.ProjectStorageAllow: + """Convert the ORM row to a ProjectStorageAllow model.""" + return models.ProjectStorageAllow( + project_id=self.project_id, + max_size=self.max_size, + updated_at=self.updated_at, + ) + + +class ProjectStorageORM(BaseORM): + """ORM model for project storage configuration.""" + + __tablename__ = "project_storage" + + id: Mapped[ULID] = mapped_column( + "id", ULIDType, primary_key=True, server_default=text("generate_ulid()"), init=False + ) + project_id: Mapped[ULID] = mapped_column( + ForeignKey(ProjectStorageAllowORM.project_id, ondelete="CASCADE"), index=True, nullable=False, unique=True + ) + """ID of the project (must exist in project_storage_allow).""" + + storage_class: Mapped[str] = mapped_column("storage_class", String(20)) + """The storage class (e.g. azurefile).""" + + size_limit: Mapped[ByteSize] = mapped_column("size_limit", ByteSizeType()) + """The storage limit in bytes.""" + + mount_path: Mapped[PurePosixPath] = mapped_column("target_path", PurePosixPathType()) + """Folder to mount to.""" + + created_by_id: Mapped[str] = mapped_column(ForeignKey(UserORM.keycloak_id), index=True, nullable=False) + """User ID of the creator of the project storage.""" + + creation_date: Mapped[datetime] = mapped_column( + "creation_date", DateTime(timezone=True), default=func.now(), nullable=False + ) + updated_at: Mapped[datetime] = mapped_column( + "updated_at", + DateTime(timezone=True), + default=None, + server_default=func.now(), + onupdate=func.now(), + nullable=False, + ) + + def dump(self) -> models.ProjectStorage: + """Convert the ORM row to a ProjectStorage model.""" + return models.ProjectStorage( + id=self.id, + project_id=self.project_id, + storage_class=self.storage_class, + size=self.size_limit, + mount_path=self.mount_path, + created_by=self.created_by_id, + creation_date=self.creation_date, + updated_at=self.updated_at, + ) diff --git a/components/renku_data_services/data_connectors/project_storage_k8s.py b/components/renku_data_services/data_connectors/project_storage_k8s.py new file mode 100644 index 000000000..82fd01692 --- /dev/null +++ b/components/renku_data_services/data_connectors/project_storage_k8s.py @@ -0,0 +1,49 @@ +"""Module for managing volumes associated to projects.""" + +from ulid import ULID + +from renku_data_services.app_config import logging +from renku_data_services.data_connectors.models import DeletedProjectStorage, ProjectStorage +from renku_data_services.k8s.models import ClusterConnection, K8sPersistentVolumeClaim +from renku_data_services.notebooks.api.classes.k8s_client import NotebookK8sClient + +logger = logging.getLogger(__name__) + + +class ProjectStorageK8s: + """Manage persistent volumes associated to a project.""" + + def __init__(self, k8s_client: NotebookK8sClient) -> None: + self.__k8s_client = k8s_client + + def __pvc_name(self, storage: ProjectStorage | DeletedProjectStorage | ULID) -> str: + if isinstance(storage, ULID): + return f"pv-{storage}-0".lower() + else: + return f"pv-{storage.project_id}-0".lower() + + async def get_or_create_volume( + self, storage: ProjectStorage, cluster: ClusterConnection + ) -> K8sPersistentVolumeClaim: + """Either creates a persistent volume or fetches an existing one.""" + name = self.__pvc_name(storage) + pvc = await self.__k8s_client.get_persistent_volume_claim(name) + if not pvc: + logger.debug(f"Create project storage for project: {storage.project_id} with name {name}") + pvc = K8sPersistentVolumeClaim.new( + cluster=cluster.id, + name=name, + namespace=cluster.namespace, + accessModes=["ReadWriteMany"], + storage_class=storage.storage_class, + size=storage.size, + labels={"renku.io/project_id": str(storage.project_id)}, + ) + await self.__k8s_client.create_persistent_volume(pvc) + return pvc + + async def delete_volume(self, project: DeletedProjectStorage | ULID) -> None: + """Delete a persistent volume associated to the project.""" + + name = self.__pvc_name(project) + await self.__k8s_client.delete_persistent_volume(name) diff --git a/components/renku_data_services/k8s/models.py b/components/renku_data_services/k8s/models.py index 40575aa0a..4b4f115ed 100644 --- a/components/renku_data_services/k8s/models.py +++ b/components/renku_data_services/k8s/models.py @@ -14,6 +14,7 @@ from kr8s.objects import Secret from kubernetes.client import V1Secret +from renku_data_services.base_models.bytesize import ByteSize from renku_data_services.errors import ProgrammingError, errors from renku_data_services.k8s.constants import DUMMY_TASK_RUN_USER_ID, ClusterId @@ -482,3 +483,58 @@ class DeletePropagationPolicy(StrEnum): foreground = "Foreground" background = "Background" + + +class K8sPersistentVolumeClaim(K8sObject): + """Represents a k8s persistent volume claim.""" + + def __init__( + self, + name: str, + namespace: str, + cluster: ClusterId, + manifest: Box, + ) -> None: + super().__init__( + name=name, + namespace=namespace, + cluster=cluster, + gvk=GVK(version="v1", kind="PersistentVolumeClaim"), + manifest=manifest, + ) + + @classmethod + def new( + cls, + name: str, + cluster: ClusterId, + namespace: str, + accessModes: list[str], + storage_class: str, + size: ByteSize, + labels: dict[str, str], + ) -> Self: + """Create a new perstistent volume claim.""" + return cls( + name=name, + namespace=namespace, + cluster=cluster, + manifest=Box( + { + "metadata": { + "namespace": namespace, + "name": name, + "labels": labels, + }, + "spec": { + "accessModes": accessModes, + "storageClassName": storage_class, + "resources": {"requests": {"storage": f"{size.to_gibi()}Gi"}}, + }, + } + ), + ) + + def get_storage_class(self) -> str: + """Return the storage class name from the manifest.""" + return str(self.manifest.spec.storageClassName) diff --git a/components/renku_data_services/migrations/versions/21c885ddf188_add_project_storage_allow_list.py b/components/renku_data_services/migrations/versions/21c885ddf188_add_project_storage_allow_list.py new file mode 100644 index 000000000..c47a36b01 --- /dev/null +++ b/components/renku_data_services/migrations/versions/21c885ddf188_add_project_storage_allow_list.py @@ -0,0 +1,70 @@ +"""add_project_storage_allow_list + +Revision ID: 21c885ddf188 +Revises: fa511acd4581 +Create Date: 2026-07-15 08:37:02.366648 + +""" + +import sqlalchemy as sa +from alembic import op + +from renku_data_services.utils.sqlalchemy import ByteSizeType, ULIDType + +# revision identifiers, used by Alembic. +revision = "21c885ddf188" +down_revision = "fa511acd4581" +branch_labels = None +depends_on = None + + +def upgrade() -> None: + # ### commands auto generated by Alembic - please adjust! ### + op.create_table( + "project_storage_allow", + sa.Column("project_id", ULIDType(), nullable=False), + sa.Column("max_size", ByteSizeType(), nullable=False), + sa.Column("updated_at", sa.DateTime(timezone=True), server_default=sa.text("now()"), nullable=False), + sa.ForeignKeyConstraint(["project_id"], ["projects.projects.id"], ondelete="CASCADE"), + sa.PrimaryKeyConstraint("project_id"), + schema="storage", + ) + op.create_index( + op.f("ix_storage_project_storage_allow_project_id"), + "project_storage_allow", + ["project_id"], + unique=True, + schema="storage", + ) + op.drop_constraint("project_storage_project_id_fkey", "project_storage", schema="storage", type_="foreignkey") + op.create_foreign_key( + "project_storage_project_id_allow_fkey", + "project_storage", + "project_storage_allow", + ["project_id"], + ["project_id"], + source_schema="storage", + referent_schema="storage", + ondelete="CASCADE", + ) + # ### end Alembic commands ### + + +def downgrade() -> None: + # ### commands auto generated by Alembic - please adjust! ### + op.drop_constraint("project_storage_project_id_allow_fkey", "project_storage", schema="storage", type_="foreignkey") + op.create_foreign_key( + "project_storage_project_id_fkey", + "project_storage", + "projects", + ["project_id"], + ["id"], + source_schema="storage", + referent_schema="projects", + ondelete="CASCADE", + ) + op.drop_index( + op.f("ix_storage_project_storage_allow_project_id"), table_name="project_storage_allow", schema="storage" + ) + op.drop_table("project_storage_allow", schema="storage") + # ### end Alembic commands ### diff --git a/components/renku_data_services/migrations/versions/fa511acd4581_add_project_storage.py b/components/renku_data_services/migrations/versions/fa511acd4581_add_project_storage.py new file mode 100644 index 000000000..5b9a5ae5d --- /dev/null +++ b/components/renku_data_services/migrations/versions/fa511acd4581_add_project_storage.py @@ -0,0 +1,59 @@ +"""add_project_storage + +Revision ID: fa511acd4581 +Revises: 01k4dy9r2we4 +Create Date: 2026-07-09 14:10:42.701383 + +""" + +import sqlalchemy as sa +from alembic import op + +from renku_data_services.utils.sqlalchemy import ByteSizeType, ULIDType + +# revision identifiers, used by Alembic. +revision = "fa511acd4581" +down_revision = "01k4dy9r2we4" +branch_labels = None +depends_on = None + + +def upgrade() -> None: + # ### commands auto generated by Alembic - please adjust! ### + op.create_table( + "project_storage", + sa.Column("id", ULIDType(), server_default=sa.text("generate_ulid()"), nullable=False), + sa.Column("project_id", ULIDType(), nullable=False), + sa.Column("storage_class", sa.String(length=20), nullable=False), + sa.Column("size_limit", ByteSizeType(), nullable=False), + sa.Column("target_path", sa.String(), nullable=False), + sa.Column("created_by_id", sa.String(length=36), nullable=False), + sa.Column("creation_date", sa.DateTime(timezone=True), nullable=False), + sa.Column("updated_at", sa.DateTime(timezone=True), server_default=sa.text("now()"), nullable=False), + sa.ForeignKeyConstraint( + ["created_by_id"], + ["users.users.keycloak_id"], + ), + sa.ForeignKeyConstraint(["project_id"], ["projects.projects.id"], ondelete="CASCADE"), + sa.PrimaryKeyConstraint("id"), + schema="storage", + ) + op.create_index( + op.f("ix_storage_project_storage_created_by_id"), + "project_storage", + ["created_by_id"], + unique=False, + schema="storage", + ) + op.create_index( + op.f("ix_storage_project_storage_project_id"), "project_storage", ["project_id"], unique=True, schema="storage" + ) + # ### end Alembic commands ### + + +def downgrade() -> None: + # ### commands auto generated by Alembic - please adjust! ### + op.drop_index(op.f("ix_storage_project_storage_project_id"), table_name="project_storage", schema="storage") + op.drop_index(op.f("ix_storage_project_storage_created_by_id"), table_name="project_storage", schema="storage") + op.drop_table("project_storage", schema="storage") + # ### end Alembic commands ### diff --git a/components/renku_data_services/notebooks/api/classes/k8s_client.py b/components/renku_data_services/notebooks/api/classes/k8s_client.py index 3d6ffe2f5..5df22eada 100644 --- a/components/renku_data_services/notebooks/api/classes/k8s_client.py +++ b/components/renku_data_services/notebooks/api/classes/k8s_client.py @@ -10,7 +10,7 @@ import httpx from box import Box from kr8s import NotFoundError, ServerError -from kr8s.asyncio.objects import Pod, Secret, StatefulSet +from kr8s.asyncio.objects import PersistentVolumeClaim, Pod, Secret, StatefulSet from renku_data_services.app_config import logging from renku_data_services.base_models import APIUser @@ -26,6 +26,7 @@ K8sObjectFilter, K8sObjectMeta, K8sPatches, + K8sPersistentVolumeClaim, K8sSecret, ) from renku_data_services.notebooks.api.classes.auth import GitlabToken, RenkuTokens @@ -34,6 +35,8 @@ from renku_data_services.notebooks.util.kubernetes_ import find_env_var from renku_data_services.notebooks.util.retries import retry_with_exponential_backoff_async +logger = logging.getLogger(__name__) + class NotebookK8sClient(SecretClient): """A K8s Client for Notebooks.""" @@ -470,3 +473,32 @@ async def create_or_patch_secret(self, secret: K8sSecret) -> K8sSecret: logger.debug(f"Patching secret {secret.namespace}/{secret.name}") result = await self.patch_secret(secret, secret.to_patch()) return result + + async def get_persistent_volume_claim(self, name: str) -> K8sPersistentVolumeClaim | None: + """Get a persistent volume claim by name.""" + + obj = await self._get(name, GVK.from_kr8s_object(PersistentVolumeClaim), None) + if not obj: + return None + + cluster = await self.__client.cluster_by_id(obj.cluster) + return K8sPersistentVolumeClaim( + name=obj.name, namespace=cluster.namespace, cluster=obj.cluster, manifest=obj.manifest + ) + + async def delete_persistent_volume(self, name: str) -> None: + """Delete a persistent volume.""" + + obj = await self.get_persistent_volume_claim(name) + if obj: + await self.__client.delete(obj) + + async def create_persistent_volume(self, pvc: K8sPersistentVolumeClaim) -> None: + """Create a persistent volume for the given project storage.""" + + if pvc.get_storage_class() != "azurefile": + raise errors.ValidationError( + message=f"Currently only azurefile is supported as a storage class, got: {pvc.get_storage_class()}" + ) + + await self.__client.create(pvc, True) diff --git a/components/renku_data_services/notebooks/blueprints.py b/components/renku_data_services/notebooks/blueprints.py index 81637887d..c9dccefca 100644 --- a/components/renku_data_services/notebooks/blueprints.py +++ b/components/renku_data_services/notebooks/blueprints.py @@ -9,6 +9,7 @@ from renku_data_services import base_models from renku_data_services.app_config import logging from renku_data_services.authn.renku import RenkuSelfTokenMint +from renku_data_services.authz.authz import Authz from renku_data_services.base_api.auth import authenticate, authenticate_2 from renku_data_services.base_api.blueprint import BlueprintFactoryResponse, CustomBlueprint from renku_data_services.base_api.misc import validate_query @@ -69,6 +70,7 @@ class NotebooksNewBP(CustomBlueprint): builds_config: BuildsConfig resource_usage_service: ResourceUsageService resource_requests_repo: ResourceRequestsRepo + authz: Authz def start(self) -> BlueprintFactoryResponse: """Start a session with the new operator.""" @@ -91,6 +93,7 @@ async def _handler( git_provider_helper=self.git_provider_helper, cluster_repo=self.cluster_repo, data_connector_secret_repo=self.data_connector_secret_repo, + data_connector_repo=self.data_connector_repo, project_repo=self.project_repo, project_session_secret_repo=self.project_session_secret_repo, rp_repo=self.rp_repo, @@ -103,6 +106,7 @@ async def _handler( builds_config=self.builds_config, internal_token_mint=self.internal_token_mint, resource_usage_service=self.resource_usage_service, + authz=self.authz, ) status = 201 if created else 200 return json(session.as_apispec().model_dump(exclude_none=True, mode="json"), status) diff --git a/components/renku_data_services/notebooks/core_sessions.py b/components/renku_data_services/notebooks/core_sessions.py index 91bff3d38..804db7c26 100644 --- a/components/renku_data_services/notebooks/core_sessions.py +++ b/components/renku_data_services/notebooks/core_sessions.py @@ -13,7 +13,10 @@ from urllib.parse import urljoin, urlparse import httpx -from kubernetes.client import V1ObjectMeta, V1Secret +from kubernetes.client import ( + V1ObjectMeta, + V1Secret, +) from sanic import Request from toml import dumps from ulid import ULID @@ -21,6 +24,8 @@ from renku_data_services.app_config import logging from renku_data_services.authn.renku import RenkuSelfTokenMint +from renku_data_services.authz.authz import Authz +from renku_data_services.authz.models import ResourceType, Scope from renku_data_services.base_models import RESET, AnonymousAPIUser, APIUser, AuthenticatedAPIUser, ResetType from renku_data_services.base_models.metrics import MetricsService from renku_data_services.crc.db import ClusterRepository, ResourcePoolRepository @@ -35,9 +40,11 @@ SessionProtocol, ) from renku_data_services.data_connectors.db import ( + DataConnectorRepository, DataConnectorSecretRepository, ) from renku_data_services.data_connectors.models import DataConnectorSecret, DataConnectorWithSecrets +from renku_data_services.data_connectors.project_storage_k8s import ProjectStorageK8s from renku_data_services.errors import ValidationError, errors from renku_data_services.k8s.models import ClusterConnection, K8sSecret, sanitizer from renku_data_services.notebooks import apispec @@ -46,6 +53,7 @@ from renku_data_services.notebooks.api.classes.image import Image from renku_data_services.notebooks.api.classes.repository import GitProvider, Repository from renku_data_services.notebooks.config import GitProviderHelperProto, NotebooksConfig +from renku_data_services.notebooks.cr_amalthea_session import PersistentVolumeClaim from renku_data_services.notebooks.crs import ( AmaltheaMetadata, AmaltheaSessionSpec, @@ -180,6 +188,47 @@ async def get_extra_containers( return SessionExtraResources(containers=conts) +async def get_project_storage( + user: APIUser, + project_storage_k8s: ProjectStorageK8s, + data_connector_repo: DataConnectorRepository, + project_id: ULID, + storage_mount: PurePosixPath, + cluster: ClusterConnection, + authz: Authz, +) -> SessionExtraResources: + """If applicable, fetch the project storage and return it as SessionExtras.""" + project_storage = await data_connector_repo.get_storage_to(user, project_id) + if not project_storage: + logger.debug(f"Project {project_id} has no project storage.") + return SessionExtraResources() + + pvc = await project_storage_k8s.get_or_create_volume(project_storage, cluster) + + logger.debug(f"Configuring project storage for {project_id}: {project_storage}") + mount_path = project_storage.mount_path + if not mount_path.is_absolute(): + mount_path = storage_mount / mount_path + + mount_name = f"ps-{project_id}-0".lower() + can_write = await authz.has_permission(user, ResourceType.project, project_id, Scope.WRITE) + return SessionExtraResources( + volume_mounts=[ + ExtraVolumeMount( + mountPath=mount_path.as_posix(), + name=mount_name, + readOnly=not can_write, + ) + ], + volumes=[ + ExtraVolume( + name=mount_name, + persistentVolumeClaim=PersistentVolumeClaim(claimName=pvc.name), + ) + ], + ) + + async def get_auth_secret_authenticated( nb_config: NotebooksConfig, user: AuthenticatedAPIUser, @@ -876,6 +925,7 @@ async def start_session( nb_config: NotebooksConfig, git_provider_helper: GitProviderHelperProto, cluster_repo: ClusterRepository, + data_connector_repo: DataConnectorRepository, data_connector_secret_repo: DataConnectorSecretRepository, project_repo: ProjectRepository, project_session_secret_repo: ProjectSessionSecretRepository, @@ -889,6 +939,7 @@ async def start_session( builds_config: BuildsConfig, internal_token_mint: RenkuSelfTokenMint, resource_usage_service: ResourceUsageService, + authz: Authz, ) -> tuple[AmaltheaSessionV1Alpha1, bool]: """Start an Amalthea session. @@ -1014,6 +1065,14 @@ async def start_session( await get_extra_containers(nb_config, server_name, user, repositories, git_providers, internal_token_mint) ) + # project storage + project_storage_k8s = ProjectStorageK8s(nb_config.k8s_v2_client) + session_extras = session_extras.concat( + await get_project_storage( + user, project_storage_k8s, data_connector_repo, project.id, storage_mount, cluster, authz + ) + ) + # Cluster settings (ingress, storage class, etc) cluster_settings: ClusterSettings try: diff --git a/components/renku_data_services/project/blueprints.py b/components/renku_data_services/project/blueprints.py index 30211825e..101fc3f4a 100644 --- a/components/renku_data_services/project/blueprints.py +++ b/components/renku_data_services/project/blueprints.py @@ -23,6 +23,7 @@ from renku_data_services.base_models.metrics import MetricsService, ProjectCreationType from renku_data_services.base_models.validation import validate_and_dump, validated_json from renku_data_services.data_connectors.db import DataConnectorRepository +from renku_data_services.data_connectors.project_storage_k8s import ProjectStorageK8s from renku_data_services.errors import errors from renku_data_services.project import apispec from renku_data_services.project import models as project_models @@ -57,6 +58,7 @@ class ProjectsBP(CustomBlueprint): session_repo: SessionRepository session_secret_repo: ProjectSessionSecretRepository metrics: MetricsService + project_storage_k8s: ProjectStorageK8s def get_all(self) -> BlueprintFactoryResponse: """List all projects.""" @@ -277,7 +279,9 @@ def delete(self) -> BlueprintFactoryResponse: @authenticate(self.authenticator) @only_authenticated async def _delete(_: Request, user: base_models.APIUser, project_id: ULID) -> HTTPResponse: - await self.project_repo.delete_project(user=user, project_id=project_id) + deleted = await self.project_repo.delete_project(user=user, project_id=project_id) + if deleted: + await self.project_storage_k8s.delete_volume(deleted.id) return HTTPResponse(status=204) return "/projects/", ["DELETE"], _delete diff --git a/components/renku_data_services/utils/core.py b/components/renku_data_services/utils/core.py index 1e373908b..c745bf841 100644 --- a/components/renku_data_services/utils/core.py +++ b/components/renku_data_services/utils/core.py @@ -82,11 +82,39 @@ async def transaction_wrapper(self: _WithSessionMaker, *args: _P.args, **kwargs: kwargs["session"] = session return await f(self, *args, **kwargs) else: - return await f(self, *args, **kwargs) + existing_session: AsyncSession = cast(AsyncSession, session_kwarg) + if existing_session.in_transaction(): + return await f(self, *args, **kwargs) + else: + async with existing_session.begin(): + return await f(self, *args, **kwargs) return transaction_wrapper +def with_db_session( + f: Callable[Concatenate[_WithSessionMaker, _P], Awaitable[_T]], +) -> Callable[Concatenate[_WithSessionMaker, _P], Awaitable[_T]]: + """Initializes a session (but not a transaction).""" + + @functools.wraps(f) + async def session_wrapper(self: _WithSessionMaker, *args: _P.args, **kwargs: _P.kwargs) -> _T: + session_kwarg = kwargs.get("session") + if "session" in kwargs and session_kwarg is not None and not isinstance(session_kwarg, AsyncSession): + raise errors.ProgrammingError( + message="The decorator that starts a DB transaction encountered an existing session " + f"in the keyword arguments but the session is of an unexpected type {type(session_kwarg)}" + ) + if session_kwarg is None: + async with self.session_maker() as session: + kwargs["session"] = session + return await f(self, *args, **kwargs) + else: + return await f(self, *args, **kwargs) + + return session_wrapper + + def _get_openbis_url(openbis_host: str) -> str: return f"https://{openbis_host}/openbis/openbis/rmi-application-server-v3.json" diff --git a/components/renku_data_services/utils/sqlalchemy.py b/components/renku_data_services/utils/sqlalchemy.py index 32241a261..d79ef35ea 100644 --- a/components/renku_data_services/utils/sqlalchemy.py +++ b/components/renku_data_services/utils/sqlalchemy.py @@ -7,6 +7,7 @@ from sqlalchemy import Dialect, types from ulid import ULID +from renku_data_services.base_models.bytesize import ByteSize from renku_data_services.resource_usage.model import ComputeCapacity, Credit, DataSize @@ -126,3 +127,18 @@ def process_result_value(self, value: str | None, dialect: Dialect) -> PurePosix if value is None: return None return PurePosixPath(value) + + +class ByteSizeType(types.TypeDecorator): + """SQLAlchemy type that maps ByteSize to a BigInteger column.""" + + impl = types.BigInteger + cache_ok = True + + def process_bind_param(self, value: ByteSize | None, dialect: Dialect) -> int | None: + """Convert ByteSize to an integer for storage.""" + return value.to_bytes() if value is not None else None + + def process_result_value(self, value: int | None, dialect: Dialect) -> ByteSize | None: + """Convert a stored integer back to ByteSize.""" + return ByteSize(value) if value is not None else None diff --git a/flake.lock b/flake.lock index bbcea675f..c693cc2ac 100644 --- a/flake.lock +++ b/flake.lock @@ -8,11 +8,11 @@ ] }, "locked": { - "lastModified": 1776238255, - "narHash": "sha256-ClPxl2WDe8PzO5hEEnhvqmU9U6je2HyNQ98smhZjd/w=", + "lastModified": 1783584226, + "narHash": "sha256-OYTWKSQeG5P1szap3LC913edVXCVGxIC+U6Pn9PlYRM=", "owner": "eikek", "repo": "devshell-tools", - "rev": "6f5ba0ead2a272e1e39bc1b956e5cf973473375d", + "rev": "e8e047df069c403f1dcf527b61a1e4224c1597d5", "type": "github" }, "original": { @@ -59,11 +59,11 @@ }, "nixpkgs": { "locked": { - "lastModified": 1757487488, - "narHash": "sha256-zwE/e7CuPJUWKdvvTCB7iunV4E/+G0lKfv4kk/5Izdg=", + "lastModified": 1783224372, + "narHash": "sha256-8i/87eeoqiGE4yOTjwSA3Eh/ziJRQEmd/unYU+K27sk=", "owner": "NixOS", "repo": "nixpkgs", - "rev": "ab0f3607a6c7486ea22229b92ed2d355f1482ee0", + "rev": "d407951447dcd00442e97087bf374aad70c04cea", "type": "github" }, "original": { diff --git a/flake.nix b/flake.nix index 89d0854f6..cd237cb5a 100644 --- a/flake.nix +++ b/flake.nix @@ -70,7 +70,7 @@ src = pkgs.fetchFromGitHub { owner = "SwissDataScienceCenter"; repo = "rclone"; - rev = "v1.71.2+renku-1"; + tag = "v1.71.2+renku-1"; sha256 = "sha256-NhPYEGPgpwe56zExrV3SiYsbKLb3/OuX+UOuezgJQ8w="; }; }); @@ -142,7 +142,7 @@ redis postgresql_16 jq - devshellToolsPkgs.openapi-docs + #devshellToolsPkgs.openapi-docs devshellToolsPkgs.solr devshellToolsPkgs.postgres-fg spicedb @@ -164,9 +164,14 @@ '') ( writeShellScriptBin "poetry-setup" '' + set -e venv_path="$(poetry env info -p)" if [ "$1" == "-c" ]; then echo "Removing virtual env at $venv_path" + if [ -z "$venv_path" ]; then + echo "No venv path found??? exiting." + exit 1 + fi rm -rf "$venv_path"/* fi poetry install @@ -281,7 +286,7 @@ SOLR_CORE = "renku-search-dev"; #AMALTHEA_SESSIONS_VERSION = "refs/heads/eikek/non-interactive-session"; - RENKU_ENV = "renku-ci-ds-1328"; + RENKU_ENV = "renku-ci-ds-1378"; K8S_NAMESPACE = "default"; KUBERNETES_NAMESPACE = "default"; DEBUG_LOGGING = "renku_data_services"; diff --git a/test/bases/renku_data_services/data_api/test_data_connectors.py b/test/bases/renku_data_services/data_api/test_data_connectors.py index 558c8a3cf..7e9479b30 100644 --- a/test/bases/renku_data_services/data_api/test_data_connectors.py +++ b/test/bases/renku_data_services/data_api/test_data_connectors.py @@ -1419,7 +1419,7 @@ async def test_delete_data_connector_after_linking( @pytest.mark.asyncio async def test_delete_project_after_linking( - sanic_client: SanicASGITestClient, create_data_connector, create_project, user_headers + sanic_client: SanicASGITestClient, create_data_connector, create_project, user_headers, cluster ) -> None: data_connector = await create_data_connector("Data connector 1") project = await create_project(sanic_client, "Project A") @@ -2665,6 +2665,685 @@ async def link_dc_project( return response +@pytest.mark.asyncio +async def test_post_storage_success( + sanic_client: SanicASGITestClient, + create_project, + admin_headers: dict[str, str], + user_headers: dict[str, str], +) -> None: + project = await create_project(sanic_client, "Test Project") + namespace = f"{project['namespace']}/{project['slug']}" + project_id = project["id"] + + payload = {"project_id": project_id, "max_size": 10} + _, response = await sanic_client.post( + "/api/data/data_connectors/storage/allow", headers=admin_headers, json=payload + ) + assert response.status_code == 201, response.text + + payload = {"namespace": namespace, "size": 10, "mount_path": "/data"} + _, response = await sanic_client.post("/api/data/data_connectors/storage", headers=user_headers, json=payload) + + assert response.status_code == 201, response.text + assert response.json is not None + storage = response.json + assert storage.get("project_id") == project["id"] + assert storage.get("size") == 10 + assert storage.get("mount_path") == "/data" + assert storage.get("created_by") == "user" + assert "ETag" in response.headers + + +@pytest.mark.asyncio +async def test_post_storage_unauthenticated_fails( + sanic_client: SanicASGITestClient, create_project, admin_headers: dict[str, str] +) -> None: + project = await create_project(sanic_client, "Test Project") + namespace = f"{project['namespace']}/{project['slug']}" + + project_id = project["id"] + payload = {"project_id": project_id, "max_size": 10} + _, response = await sanic_client.post( + "/api/data/data_connectors/storage/allow", headers=admin_headers, json=payload + ) + assert response.status_code == 201, response.text + + payload = {"namespace": namespace, "size": 10, "mount_path": "/data"} + _, response = await sanic_client.post("/api/data/data_connectors/storage", json=payload) + + assert response.status_code == 401, response.text + + +@pytest.mark.asyncio +async def test_post_storage_not_allowed_fails( + sanic_client: SanicASGITestClient, create_project, user_headers: dict[str, str] +) -> None: + project = await create_project(sanic_client, "Test Project") + namespace = f"{project['namespace']}/{project['slug']}" + + payload = {"namespace": namespace, "size": 10, "mount_path": "/data"} + _, response = await sanic_client.post("/api/data/data_connectors/storage", json=payload, headers=user_headers) + + assert response.status_code == 403, response.text + + +@pytest.mark.asyncio +async def test_post_storage_duplicate_fails( + sanic_client: SanicASGITestClient, create_project, user_headers: dict[str, str], admin_headers: dict[str, str] +) -> None: + project = await create_project(sanic_client, "Test Project") + namespace = f"{project['namespace']}/{project['slug']}" + + project_id = project["id"] + payload = {"project_id": project_id, "max_size": 10} + _, response = await sanic_client.post( + "/api/data/data_connectors/storage/allow", headers=admin_headers, json=payload + ) + assert response.status == 201 + + payload = {"namespace": namespace, "size": 10, "mount_path": "/data"} + _, response = await sanic_client.post("/api/data/data_connectors/storage", headers=user_headers, json=payload) + assert response.status_code == 201, response.text + + _, response = await sanic_client.post("/api/data/data_connectors/storage", headers=user_headers, json=payload) + + assert response.status_code == 422, response.text + + +@pytest.mark.asyncio +async def test_get_one_storage_success( + sanic_client: SanicASGITestClient, create_project, user_headers: dict[str, str], admin_headers: dict[str, str] +) -> None: + project = await create_project(sanic_client, "Test Project") + namespace = f"{project['namespace']}/{project['slug']}" + + project_id = project["id"] + payload = {"project_id": project_id, "max_size": 10} + _, response = await sanic_client.post( + "/api/data/data_connectors/storage/allow", headers=admin_headers, json=payload + ) + assert response.status_code == 201, response.text + + payload = {"namespace": namespace, "size": 10, "mount_path": "/data"} + _, response = await sanic_client.post("/api/data/data_connectors/storage", headers=user_headers, json=payload) + assert response.status_code == 201, response.text + storage_id = response.json["id"] + + _, response = await sanic_client.get(f"/api/data/data_connectors/storage/{storage_id}", headers=user_headers) + + assert response.status_code == 200, response.text + assert response.json is not None + storage = response.json + assert storage.get("id") == storage_id + assert storage.get("project_id") == project["id"] + assert storage.get("size") == 10 + assert storage.get("mount_path") == "/data" + + +@pytest.mark.asyncio +async def test_get_one_storage_not_found(sanic_client: SanicASGITestClient, user_headers: dict[str, str]) -> None: + from ulid import ULID + + non_existent_id = str(ULID()) + _, response = await sanic_client.get(f"/api/data/data_connectors/storage/{non_existent_id}", headers=user_headers) + + assert response.status_code == 404, response.text + + +@pytest.mark.asyncio +async def test_get_one_storage_etag( + sanic_client: SanicASGITestClient, create_project, user_headers: dict[str, str], admin_headers: dict[str, str] +) -> None: + project = await create_project(sanic_client, "Test Project") + namespace = f"{project['namespace']}/{project['slug']}" + + project_id = project["id"] + payload = {"project_id": project_id, "max_size": 10} + _, response = await sanic_client.post( + "/api/data/data_connectors/storage/allow", headers=admin_headers, json=payload + ) + assert response.status_code == 201, response.text + + payload = {"namespace": namespace, "size": 10, "mount_path": "/data"} + _, response = await sanic_client.post("/api/data/data_connectors/storage", headers=user_headers, json=payload) + assert response.status_code == 201, response.text + storage_id = response.json["id"] + etag = response.headers["ETag"] + + headers = merge_headers(user_headers, {"If-None-Match": etag}) + _, response = await sanic_client.get(f"/api/data/data_connectors/storage/{storage_id}", headers=headers) + + assert response.status_code == 304, response.text + + +@pytest.mark.asyncio +async def test_get_storage_to_project_for_no_storage( + sanic_client: SanicASGITestClient, create_project, user_headers: dict[str, str] +) -> None: + project = await create_project(sanic_client, "Test Project") + + _, response = await sanic_client.get(f"/api/data/projects/{project['id']}/storage", headers=user_headers) + + assert response.status_code == 200, response.text + assert response.json == [] + + +@pytest.mark.asyncio +async def test_get_storage_to_project_success( + sanic_client: SanicASGITestClient, create_project, user_headers: dict[str, str], admin_headers: dict[str, str] +) -> None: + project = await create_project(sanic_client, "Test Project") + namespace = f"{project['namespace']}/{project['slug']}" + + project_id = project["id"] + payload = {"project_id": project_id, "max_size": 10} + _, response = await sanic_client.post( + "/api/data/data_connectors/storage/allow", headers=admin_headers, json=payload + ) + assert response.status == 201 + + payload = {"namespace": namespace, "size": 10, "mount_path": "/data"} + _, response = await sanic_client.post("/api/data/data_connectors/storage", headers=user_headers, json=payload) + assert response.status_code == 201, response.text + + _, response = await sanic_client.get(f"/api/data/projects/{project['id']}/storage", headers=user_headers) + + assert response.status_code == 200, response.text + assert len(response.json) == 1 + storage = response.json[0] + assert storage.get("project_id") == project["id"] + assert storage.get("size") == 10 + assert storage.get("mount_path") == "/data" + + +@pytest.mark.asyncio +async def test_delete_storage_success( + sanic_client: SanicASGITestClient, + create_project, + user_headers: dict[str, str], + admin_headers: dict[str, str], + cluster, +) -> None: + project = await create_project(sanic_client, "Test Project") + namespace = f"{project['namespace']}/{project['slug']}" + + project_id = project["id"] + payload = {"project_id": project_id, "max_size": 10} + _, response = await sanic_client.post( + "/api/data/data_connectors/storage/allow", headers=admin_headers, json=payload + ) + assert response.status_code == 201, response.text + + payload = {"namespace": namespace, "size": 10, "mount_path": "/data"} + _, response = await sanic_client.post("/api/data/data_connectors/storage", headers=user_headers, json=payload) + assert response.status_code == 201, response.text + storage_id = response.json["id"] + + _, response = await sanic_client.delete(f"/api/data/data_connectors/storage/{storage_id}", headers=user_headers) + assert response.status_code == 204, response.text + + _, response = await sanic_client.get(f"/api/data/projects/{project['id']}/storage", headers=user_headers) + assert response.status_code == 200, response.text + assert response.json == [] + + +@pytest.mark.asyncio +async def test_delete_storage_unauthenticated_fails( + sanic_client: SanicASGITestClient, create_project, user_headers: dict[str, str], admin_headers: dict[str, str] +) -> None: + project = await create_project(sanic_client, "Test Project") + namespace = f"{project['namespace']}/{project['slug']}" + + project_id = project["id"] + payload = {"project_id": project_id, "max_size": 10} + _, response = await sanic_client.post( + "/api/data/data_connectors/storage/allow", headers=admin_headers, json=payload + ) + assert response.status_code == 201, response.text + + payload = {"namespace": namespace, "size": 10, "mount_path": "/data"} + _, response = await sanic_client.post("/api/data/data_connectors/storage", headers=user_headers, json=payload) + assert response.status_code == 201, response.text + storage_id = response.json["id"] + + _, response = await sanic_client.delete(f"/api/data/data_connectors/storage/{storage_id}") + + assert response.status_code == 401, response.text + + +@pytest.mark.asyncio +async def test_post_storage_allow_success( + sanic_client: SanicASGITestClient, create_project, admin_headers: dict[str, str] +) -> None: + project = await create_project(sanic_client, "Test Project") + project_id = project["id"] + + payload = {"project_id": project_id, "max_size": 10} + _, response = await sanic_client.post( + "/api/data/data_connectors/storage/allow", headers=admin_headers, json=payload + ) + + assert response.status_code == 201, response.text + assert response.json is not None + allow = response.json + assert allow.get("project_id") == project_id + assert allow.get("max_size") == 10 + + +@pytest.mark.asyncio +async def test_post_storage_allow_requires_admin( + sanic_client: SanicASGITestClient, create_project, user_headers: dict[str, str] +) -> None: + project = await create_project(sanic_client, "Test Project") + + payload = {"project_id": project["id"], "max_size": 10} + _, response = await sanic_client.post("/api/data/data_connectors/storage/allow", headers=user_headers, json=payload) + + assert response.status_code == 403, response.text + + +@pytest.mark.asyncio +async def test_post_storage_allow_unauthenticated_fails( + sanic_client: SanicASGITestClient, create_project, user_headers: dict[str, str] +) -> None: + project = await create_project(sanic_client, "Test Project") + + payload = {"project_id": project["id"], "max_size": 10} + _, response = await sanic_client.post("/api/data/data_connectors/storage/allow", json=payload) + + assert response.status_code == 401, response.text + + +@pytest.mark.asyncio +async def test_post_storage_allow_duplicate_fails( + sanic_client: SanicASGITestClient, create_project, admin_headers: dict[str, str] +) -> None: + project = await create_project(sanic_client, "Test Project") + project_id = project["id"] + + payload = {"project_id": project_id, "max_size": 10} + _, response = await sanic_client.post( + "/api/data/data_connectors/storage/allow", headers=admin_headers, json=payload + ) + assert response.status_code == 201, response.text + + _, response = await sanic_client.post( + "/api/data/data_connectors/storage/allow", headers=admin_headers, json=payload + ) + + assert response.status_code == 422, response.text + + +@pytest.mark.asyncio +async def test_delete_storage_allow_success( + sanic_client: SanicASGITestClient, create_project, admin_headers: dict[str, str] +) -> None: + project = await create_project(sanic_client, "Test Project") + project_id = project["id"] + + payload = {"project_id": project_id, "max_size": 10} + _, response = await sanic_client.post( + "/api/data/data_connectors/storage/allow", headers=admin_headers, json=payload + ) + assert response.status_code == 201, response.text + + _, response = await sanic_client.delete( + f"/api/data/data_connectors/storage/allow/{project_id}", headers=admin_headers + ) + + assert response.status_code == 204, response.text + + # Re-adding after deletion should succeed + _, response = await sanic_client.post( + "/api/data/data_connectors/storage/allow", headers=admin_headers, json=payload + ) + assert response.status_code == 201, response.text + + +@pytest.mark.asyncio +async def test_delete_storage_allow_requires_admin( + sanic_client: SanicASGITestClient, create_project, admin_headers: dict[str, str], user_headers: dict[str, str] +) -> None: + project = await create_project(sanic_client, "Test Project") + project_id = project["id"] + + payload = {"project_id": project_id, "max_size": 10} + _, response = await sanic_client.post( + "/api/data/data_connectors/storage/allow", headers=admin_headers, json=payload + ) + assert response.status_code == 201, response.text + + _, response = await sanic_client.delete( + f"/api/data/data_connectors/storage/allow/{project_id}", headers=user_headers + ) + + assert response.status_code == 403, response.text + + +@pytest.mark.asyncio +async def test_get_storage_allow_success( + sanic_client: SanicASGITestClient, create_project, admin_headers: dict[str, str], user_headers: dict[str, str] +) -> None: + project = await create_project(sanic_client, "Test Project") + project_id = project["id"] + + payload = {"project_id": project_id, "max_size": 10} + _, response = await sanic_client.post( + "/api/data/data_connectors/storage/allow", headers=admin_headers, json=payload + ) + assert response.status_code == 201, response.text + + _, response = await sanic_client.get(f"/api/data/data_connectors/storage/allow/{project_id}", headers=user_headers) + + assert response.status_code == 200, response.text + assert response.json is not None + assert response.json.get("project_id") == project_id + assert response.json.get("max_size") == 10 + + +@pytest.mark.asyncio +async def test_get_storage_allow_not_in_list( + sanic_client: SanicASGITestClient, create_project, user_headers: dict[str, str] +) -> None: + project = await create_project(sanic_client, "Test Project") + project_id = project["id"] + + _, response = await sanic_client.get(f"/api/data/data_connectors/storage/allow/{project_id}", headers=user_headers) + + assert response.status_code == 404, response.text + + +@pytest.mark.asyncio +async def test_get_storage_allow_unauthenticated( + sanic_client: SanicASGITestClient, create_project, admin_headers: dict[str, str] +) -> None: + project = await create_project(sanic_client, "Test Project") + project_id = project["id"] + + payload = {"project_id": project_id, "max_size": 10} + _, response = await sanic_client.post( + "/api/data/data_connectors/storage/allow", headers=admin_headers, json=payload + ) + assert response.status_code == 201, response.text + + _, response = await sanic_client.get(f"/api/data/data_connectors/storage/allow/{project_id}") + + assert response.status_code == 401, response.text + + +@pytest.mark.asyncio +async def test_patch_storage_success( + sanic_client: SanicASGITestClient, create_project, user_headers: dict[str, str], admin_headers: dict[str, str] +) -> None: + project = await create_project(sanic_client, "Test Project") + namespace = f"{project['namespace']}/{project['slug']}" + + project_id = project["id"] + payload = {"project_id": project_id, "max_size": 10} + _, response = await sanic_client.post( + "/api/data/data_connectors/storage/allow", headers=admin_headers, json=payload + ) + assert response.status_code == 201, response.text + + payload = {"namespace": namespace, "size": 5, "mount_path": "/data"} + _, response = await sanic_client.post("/api/data/data_connectors/storage", headers=user_headers, json=payload) + assert response.status_code == 201, response.text + storage = response.json + storage_id = storage["id"] + original_etag = response.headers["ETag"] + + # Patch the size + headers = merge_headers(user_headers, {"If-Match": original_etag}) + patch = {"size": 8} + _, response = await sanic_client.patch( + f"/api/data/data_connectors/storage/{storage_id}", headers=headers, json=patch + ) + + assert response.status_code == 200, response.text + assert response.json is not None + updated_storage = response.json + assert updated_storage.get("id") == storage_id + assert updated_storage.get("size") == 8 + assert updated_storage.get("mount_path") == "/data" + + +@pytest.mark.asyncio +async def test_patch_storage_mount_path( + sanic_client: SanicASGITestClient, create_project, user_headers: dict[str, str], admin_headers: dict[str, str] +) -> None: + project = await create_project(sanic_client, "Test Project") + namespace = f"{project['namespace']}/{project['slug']}" + + project_id = project["id"] + payload = {"project_id": project_id, "max_size": 10} + _, response = await sanic_client.post( + "/api/data/data_connectors/storage/allow", headers=admin_headers, json=payload + ) + assert response.status_code == 201, response.text + + payload = {"namespace": namespace, "size": 10, "mount_path": "/data"} + _, response = await sanic_client.post("/api/data/data_connectors/storage", headers=user_headers, json=payload) + assert response.status_code == 201, response.text + storage = response.json + storage_id = storage["id"] + + # Patch the mount path + headers = merge_headers(user_headers, {"If-Match": response.headers["ETag"]}) + patch = {"mount_path": "/new/mount"} + _, response = await sanic_client.patch( + f"/api/data/data_connectors/storage/{storage_id}", headers=headers, json=patch + ) + + assert response.status_code == 200, response.text + assert response.json is not None + updated_storage = response.json + assert updated_storage.get("id") == storage_id + assert updated_storage.get("size") == 10 + assert updated_storage.get("mount_path") == "/new/mount" + + +@pytest.mark.asyncio +async def test_patch_storage_both_fields( + sanic_client: SanicASGITestClient, create_project, user_headers: dict[str, str], admin_headers: dict[str, str] +) -> None: + project = await create_project(sanic_client, "Test Project") + namespace = f"{project['namespace']}/{project['slug']}" + + project_id = project["id"] + payload = {"project_id": project_id, "max_size": 10} + _, response = await sanic_client.post( + "/api/data/data_connectors/storage/allow", headers=admin_headers, json=payload + ) + assert response.status_code == 201, response.text + + payload = {"namespace": namespace, "size": 5, "mount_path": "/data"} + _, response = await sanic_client.post("/api/data/data_connectors/storage", headers=user_headers, json=payload) + assert response.status_code == 201, response.text + storage = response.json + storage_id = storage["id"] + + # Patch both fields at once + headers = merge_headers(user_headers, {"If-Match": response.headers["ETag"]}) + patch = {"size": 8, "mount_path": "/new/mount"} + _, response = await sanic_client.patch( + f"/api/data/data_connectors/storage/{storage_id}", headers=headers, json=patch + ) + + assert response.status_code == 200, response.text + assert response.json is not None + updated_storage = response.json + assert updated_storage.get("id") == storage_id + assert updated_storage.get("size") == 8 + assert updated_storage.get("mount_path") == "/new/mount" + + +@pytest.mark.asyncio +async def test_patch_storage_without_if_match_header( + sanic_client: SanicASGITestClient, create_project, user_headers: dict[str, str], admin_headers: dict[str, str] +) -> None: + project = await create_project(sanic_client, "Test Project") + namespace = f"{project['namespace']}/{project['slug']}" + + project_id = project["id"] + payload = {"project_id": project_id, "max_size": 10} + _, response = await sanic_client.post( + "/api/data/data_connectors/storage/allow", headers=admin_headers, json=payload + ) + assert response.status_code == 201, response.text + + payload = {"namespace": namespace, "size": 10, "mount_path": "/data"} + _, response = await sanic_client.post("/api/data/data_connectors/storage", headers=user_headers, json=payload) + assert response.status_code == 201, response.text + storage_id = response.json["id"] + + # Patch without If-Match header + patch = {"size": 20} + _, response = await sanic_client.patch( + f"/api/data/data_connectors/storage/{storage_id}", headers=user_headers, json=patch + ) + + assert response.status_code == 428, response.text + assert "If-Match header not provided" in response.text + + +@pytest.mark.asyncio +async def test_patch_storage_with_invalid_etag( + sanic_client: SanicASGITestClient, create_project, user_headers: dict[str, str], admin_headers: dict[str, str] +) -> None: + project = await create_project(sanic_client, "Test Project") + namespace = f"{project['namespace']}/{project['slug']}" + + project_id = project["id"] + payload = {"project_id": project_id, "max_size": 10} + _, response = await sanic_client.post( + "/api/data/data_connectors/storage/allow", headers=admin_headers, json=payload + ) + assert response.status_code == 201, response.text + + payload = {"namespace": namespace, "size": 5, "mount_path": "/data"} + _, response = await sanic_client.post("/api/data/data_connectors/storage", headers=user_headers, json=payload) + assert response.status_code == 201, response.text + storage_id = response.json["id"] + correct_etag = response.headers["ETag"] + + # Patch with wrong ETag + headers = merge_headers(user_headers, {"If-Match": "wrong-etag"}) + patch = {"size": 8} + _, response = await sanic_client.patch( + f"/api/data/data_connectors/storage/{storage_id}", headers=headers, json=patch + ) + + assert response.status_code == 409, response.text + + # Verify the etag changed after a successful patch + headers = merge_headers(user_headers, {"If-Match": correct_etag}) + patch = {"size": 6} + _, response = await sanic_client.patch( + f"/api/data/data_connectors/storage/{storage_id}", headers=headers, json=patch + ) + assert response.status_code == 200, response.text + new_etag = response.headers["ETag"] + assert new_etag != correct_etag + + +@pytest.mark.asyncio +async def test_patch_storage_not_found(sanic_client: SanicASGITestClient, user_headers: dict[str, str]) -> None: + from ulid import ULID + + non_existent_id = str(ULID()) + headers = merge_headers(user_headers, {"If-Match": "some-etag"}) + patch = {"size": 20} + _, response = await sanic_client.patch( + f"/api/data/data_connectors/storage/{non_existent_id}", headers=headers, json=patch + ) + + assert response.status_code == 404, response.text + + +@pytest.mark.asyncio +async def test_patch_storage_unauthenticated_fails( + sanic_client: SanicASGITestClient, create_project, admin_headers: dict[str, str], user_headers: dict[str, str] +) -> None: + project = await create_project(sanic_client, "Test Project") + namespace = f"{project['namespace']}/{project['slug']}" + + project_id = project["id"] + payload = {"project_id": project_id, "max_size": 10} + _, response = await sanic_client.post( + "/api/data/data_connectors/storage/allow", headers=admin_headers, json=payload + ) + assert response.status_code == 201, response.text + + payload = {"namespace": namespace, "size": 5, "mount_path": "/data"} + _, response = await sanic_client.post("/api/data/data_connectors/storage", headers=user_headers, json=payload) + assert response.status_code == 201, response.text + storage_id = response.json["id"] + + # Patch without authentication + _, response = await sanic_client.patch(f"/api/data/data_connectors/storage/{storage_id}", json={"size": 8}) + + assert response.status_code == 401, response.text + + +@pytest.mark.asyncio +async def test_patch_storage_exceeds_max_size( + sanic_client: SanicASGITestClient, create_project, user_headers: dict[str, str], admin_headers: dict[str, str] +) -> None: + project = await create_project(sanic_client, "Test Project") + namespace = f"{project['namespace']}/{project['slug']}" + + project_id = project["id"] + payload = {"project_id": project_id, "max_size": 10} + _, response = await sanic_client.post( + "/api/data/data_connectors/storage/allow", headers=admin_headers, json=payload + ) + assert response.status_code == 201, response.text + + payload = {"namespace": namespace, "size": 5, "mount_path": "/data"} + _, response = await sanic_client.post("/api/data/data_connectors/storage", headers=user_headers, json=payload) + assert response.status_code == 201, response.text + storage_id = response.json["id"] + + # Try to patch size beyond the allowed max (10GB) + headers = merge_headers(user_headers, {"If-Match": response.headers["ETag"]}) + patch = {"size": 11} + _, response = await sanic_client.patch( + f"/api/data/data_connectors/storage/{storage_id}", headers=headers, json=patch + ) + + assert response.status_code == 422, response.text + + +@pytest.mark.asyncio +async def test_patch_storage_invalid_mount_path( + sanic_client: SanicASGITestClient, create_project, user_headers: dict[str, str], admin_headers: dict[str, str] +) -> None: + project = await create_project(sanic_client, "Test Project") + namespace = f"{project['namespace']}/{project['slug']}" + + project_id = project["id"] + payload = {"project_id": project_id, "max_size": 10} + _, response = await sanic_client.post( + "/api/data/data_connectors/storage/allow", headers=admin_headers, json=payload + ) + assert response.status_code == 201, response.text + + payload = {"namespace": namespace, "size": 10, "mount_path": "/data"} + _, response = await sanic_client.post("/api/data/data_connectors/storage", headers=user_headers, json=payload) + assert response.status_code == 201, response.text + storage_id = response.json["id"] + + # Try to patch with invalid mount path + headers = merge_headers(user_headers, {"If-Match": response.headers["ETag"]}) + patch = {"mount_path": "/etc/passwd"} + _, response = await sanic_client.patch( + f"/api/data/data_connectors/storage/{storage_id}", headers=headers, json=patch + ) + + assert response.status_code == 422, response.text + + async def test_get_all_dc_links( sanic_client: SanicASGITestClient, regular_user: UserInfo, @@ -2708,3 +3387,215 @@ async def test_get_all_dc_links( assert res.status_code == 200, res.text assert len(res.json) == 2 assert p2["id"] not in [i["id"] for i in res.json] + + +# --- PATCH Project Storage Allow Tests --- + + +@pytest.mark.asyncio +async def test_patch_storage_allow_success( + sanic_client: SanicASGITestClient, create_project, admin_headers: dict[str, str], user_headers: dict[str, str] +) -> None: + project = await create_project(sanic_client, "Test Project") + project_id = project["id"] + + payload = {"project_id": project_id, "max_size": 10} + _, response = await sanic_client.post( + "/api/data/data_connectors/storage/allow", headers=admin_headers, json=payload + ) + assert response.status_code == 201, response.text + + # Get the allow entry to retrieve the etag (use user_headers as admin may not have read access) + _, response = await sanic_client.get(f"/api/data/data_connectors/storage/allow/{project_id}", headers=user_headers) + assert response.status_code == 200, response.text + etag = response.headers["ETag"] + + # Patch the max_size + headers = merge_headers(admin_headers, {"If-Match": etag}) + patch = {"max_size": 20} + _, response = await sanic_client.patch( + f"/api/data/data_connectors/storage/allow/{project_id}", headers=headers, json=patch + ) + + assert response.status_code == 200, response.text + assert response.json is not None + updated_allow = response.json + assert updated_allow.get("project_id") == project_id + assert updated_allow.get("max_size") == 20 + + +@pytest.mark.asyncio +async def test_patch_storage_allow_requires_admin( + sanic_client: SanicASGITestClient, create_project, admin_headers: dict[str, str], user_headers: dict[str, str] +) -> None: + project = await create_project(sanic_client, "Test Project") + project_id = project["id"] + + payload = {"project_id": project_id, "max_size": 10} + _, response = await sanic_client.post( + "/api/data/data_connectors/storage/allow", headers=admin_headers, json=payload + ) + assert response.status_code == 201, response.text + + # Get the allow entry to retrieve the etag (use user_headers as admin may not have read access) + _, response = await sanic_client.get(f"/api/data/data_connectors/storage/allow/{project_id}", headers=admin_headers) + assert response.status_code == 200, response.text + etag = response.headers["ETag"] + + # Try to patch as non-admin + headers = merge_headers(user_headers, {"If-Match": etag}) + patch = {"max_size": 20} + _, response = await sanic_client.patch( + f"/api/data/data_connectors/storage/allow/{project_id}", headers=headers, json=patch + ) + + assert response.status_code == 403, response.text + + +@pytest.mark.asyncio +async def test_patch_storage_allow_without_if_match_header( + sanic_client: SanicASGITestClient, create_project, admin_headers: dict[str, str] +) -> None: + project = await create_project(sanic_client, "Test Project") + project_id = project["id"] + + payload = {"project_id": project_id, "max_size": 10} + _, response = await sanic_client.post( + "/api/data/data_connectors/storage/allow", headers=admin_headers, json=payload + ) + assert response.status_code == 201, response.text + + # Patch without If-Match header + patch = {"max_size": 20} + _, response = await sanic_client.patch( + f"/api/data/data_connectors/storage/allow/{project_id}", headers=admin_headers, json=patch + ) + + assert response.status_code == 428, response.text + assert "If-Match header not provided" in response.text + + +@pytest.mark.asyncio +async def test_patch_storage_allow_with_invalid_etag( + sanic_client: SanicASGITestClient, create_project, admin_headers: dict[str, str], user_headers: dict[str, str] +) -> None: + project = await create_project(sanic_client, "Test Project") + project_id = project["id"] + + payload = {"project_id": project_id, "max_size": 10} + _, response = await sanic_client.post( + "/api/data/data_connectors/storage/allow", headers=admin_headers, json=payload + ) + assert response.status_code == 201, response.text + + # Get the allow entry to retrieve the etag (use user_headers as admin may not have read access) + _, response = await sanic_client.get(f"/api/data/data_connectors/storage/allow/{project_id}", headers=user_headers) + assert response.status_code == 200, response.text + correct_etag = response.headers["ETag"] + + # Patch with wrong ETag + headers = merge_headers(admin_headers, {"If-Match": "wrong-etag"}) + patch = {"max_size": 20} + _, response = await sanic_client.patch( + f"/api/data/data_connectors/storage/allow/{project_id}", headers=headers, json=patch + ) + + assert response.status_code == 409, response.text + + # Verify the etag changed after a successful patch + headers = merge_headers(admin_headers, {"If-Match": correct_etag}) + patch = {"max_size": 15} + _, response = await sanic_client.patch( + f"/api/data/data_connectors/storage/allow/{project_id}", headers=headers, json=patch + ) + assert response.status_code == 200, response.text + new_etag = response.headers["ETag"] + assert new_etag != correct_etag + + +@pytest.mark.asyncio +async def test_patch_storage_allow_not_in_list( + sanic_client: SanicASGITestClient, create_project, admin_headers: dict[str, str] +) -> None: + project = await create_project(sanic_client, "Test Project") + project_id = project["id"] + + headers = merge_headers(admin_headers, {"If-Match": "some-etag"}) + patch = {"max_size": 20} + _, response = await sanic_client.patch( + f"/api/data/data_connectors/storage/allow/{project_id}", headers=headers, json=patch + ) + + assert response.status_code == 404, response.text + + +@pytest.mark.asyncio +async def test_patch_storage_allow_unauthenticated_fails( + sanic_client: SanicASGITestClient, create_project, admin_headers: dict[str, str] +) -> None: + project = await create_project(sanic_client, "Test Project") + project_id = project["id"] + + payload = {"project_id": project_id, "max_size": 10} + _, response = await sanic_client.post( + "/api/data/data_connectors/storage/allow", headers=admin_headers, json=payload + ) + assert response.status_code == 201, response.text + + # Patch without authentication + _, response = await sanic_client.patch( + f"/api/data/data_connectors/storage/allow/{project_id}", json={"max_size": 20} + ) + + assert response.status_code == 401, response.text + + +@pytest.mark.asyncio +async def test_patch_storage_allow_min_size( + sanic_client: SanicASGITestClient, create_project, admin_headers: dict[str, str], user_headers: dict[str, str] +) -> None: + project = await create_project(sanic_client, "Test Project") + project_id = project["id"] + + payload = {"project_id": project_id, "max_size": 10} + _, response = await sanic_client.post( + "/api/data/data_connectors/storage/allow", headers=admin_headers, json=payload + ) + assert response.status_code == 201, response.text + + # Get the allow entry to retrieve the etag (use user_headers as admin may not have read access) + _, response = await sanic_client.get(f"/api/data/data_connectors/storage/allow/{project_id}", headers=user_headers) + assert response.status_code == 200, response.text + etag = response.headers["ETag"] + + # Try to set max_size below minimum (1GB) + headers = merge_headers(admin_headers, {"If-Match": etag}) + patch = {"max_size": 0} + _, response = await sanic_client.patch( + f"/api/data/data_connectors/storage/allow/{project_id}", headers=headers, json=patch + ) + + assert response.status_code == 422, response.text + assert "at least 1GB" in response.text + + +@pytest.mark.asyncio +async def test_get_all_storage_allow( + sanic_client: SanicASGITestClient, create_project, admin_headers: dict[str, str], user_headers: dict[str, str] +) -> None: + project = await create_project(sanic_client, "Test Project") + project_id = project["id"] + + payload = {"project_id": project_id, "max_size": 10} + _, response = await sanic_client.post( + "/api/data/data_connectors/storage/allow", headers=admin_headers, json=payload + ) + assert response.status_code == 201, response.text + + _, response = await sanic_client.get("/api/data/data_connectors/storage/allow", headers=admin_headers) + assert response.status_code == 200, response.text + assert response.json is not None, f"No json response body: {response.text}" + assert isinstance(response.json, list) + assert len(response.json) == 1 + assert response.json[0]["project_id"] == project_id + assert response.json[0]["max_size"] == 10 diff --git a/test/bases/renku_data_services/data_api/test_projects.py b/test/bases/renku_data_services/data_api/test_projects.py index bb3a840eb..49a9100b9 100644 --- a/test/bases/renku_data_services/data_api/test_projects.py +++ b/test/bases/renku_data_services/data_api/test_projects.py @@ -358,7 +358,7 @@ async def test_result_is_sorted_by_creation_date(create_project, sanic_client, u @pytest.mark.asyncio -async def test_delete_project(create_project, sanic_client, user_headers, app_manager) -> None: +async def test_delete_project(create_project, sanic_client, user_headers, app_manager, cluster) -> None: await app_manager.search_updates_repo.clear_all() # Create some projects diff --git a/test/components/renku_data_services/search/test_reprovision.py b/test/components/renku_data_services/search/test_reprovision.py index 96cb03080..e956bf2fc 100644 --- a/test/components/renku_data_services/search/test_reprovision.py +++ b/test/components/renku_data_services/search/test_reprovision.py @@ -10,6 +10,7 @@ from renku_data_services.authz.models import Visibility from renku_data_services.base_models.core import APIUser, NamespacePath from renku_data_services.base_models.metrics import MetricsService +from renku_data_services.data_connectors.config import ProjectStorageConfig from renku_data_services.data_connectors.db import DataConnectorRepository from renku_data_services.data_connectors.models import ( CloudStorageCore, @@ -54,7 +55,7 @@ def make_setup(app_manager_instance, solr_config) -> Setup: gr = GroupRepository(sess, authz, search_updates) ur = UserRepo(sess, gr, search_updates, None, MagicMock(spec=MetricsService), authz) pr = ProjectRepository(sess, gr, search_updates, authz) - dcr = DataConnectorRepository(sess, authz, pr, gr, search_updates) + dcr = DataConnectorRepository(sess, authz, pr, gr, search_updates, ProjectStorageConfig.from_env()) sr = SearchReprovision( search_updates_repo=search_updates, reprovisioning_repo=ReprovisioningRepository(sess), diff --git a/test/utils.py b/test/utils.py index 0ad39e445..1c0fde1c8 100644 --- a/test/utils.py +++ b/test/utils.py @@ -37,6 +37,7 @@ from renku_data_services.data_connectors.db import DataConnectorRepository, DataConnectorSecretRepository from renku_data_services.data_connectors.deposits.envidat import EnvidatClient from renku_data_services.data_connectors.deposits.zenodo import ZenodoAPIClient +from renku_data_services.data_connectors.project_storage_k8s import ProjectStorageK8s from renku_data_services.db_config.config import DBConfig from renku_data_services.git.gitlab import DummyGitlabAPI from renku_data_services.k8s.clients import ( @@ -324,6 +325,7 @@ def from_env( project_repo=project_repo, group_repo=group_repo, search_updates_repo=search_updates_repo, + project_storage_config=config.project_storage_config, ) data_connector_secret_repo = DataConnectorSecretRepository( session_maker=config.db.async_session_maker, @@ -364,6 +366,7 @@ def from_env( occurrence_repo = OccurrenceRepository(session_maker=config.db.async_session_maker) resource_requests_repo = ResourceRequestsRepo(session_maker=config.db.async_session_maker) resource_usage_service = ResourceUsageService(resource_requests_repo) + project_storage_k8s = ProjectStorageK8s(config.nb_config.k8s_v2_client) return cls( config=config, @@ -416,6 +419,7 @@ def from_env( secret_client=secret_client, internal_token_mint=internal_token_mint, internal_scope_verifier=internal_scope_verifier, + project_storage_k8s=project_storage_k8s, ) def __post_init__(self) -> None: