diff --git a/components/renku_data_services/base_api/auth.py b/components/renku_data_services/base_api/auth.py index f468296f4..16b76b09d 100644 --- a/components/renku_data_services/base_api/auth.py +++ b/components/renku_data_services/base_api/auth.py @@ -71,30 +71,6 @@ async def decorated_function(request: Request, *args: _P.args, **kwargs: _P.kwar return decorator -def validate_path_project_id( - f: Callable[Concatenate[Request, _P], Coroutine[Any, Any, _T]], -) -> Callable[Concatenate[Request, _P], Coroutine[Any, Any, _T]]: - """Decorator for a Sanic handler that validates the project_id path parameter.""" - _path_project_id_regex = re.compile(r"^[A-Za-z0-9]{26}$") - - @wraps(f) - async def decorated_function(request: Request, *args: _P.args, **kwargs: _P.kwargs) -> _T: - project_id = cast(str | None, kwargs.get("project_id")) - if not project_id: - raise errors.ProgrammingError( - message="Could not find 'project_id' in the keyword arguments for the handler in order to validate it." - ) - if not _path_project_id_regex.match(project_id): - raise errors.ValidationError( - message=f"The 'project_id' path parameter {project_id} does not match the required " - f"regex {_path_project_id_regex}" - ) - - return await f(request, *args, **kwargs) - - return decorated_function - - def validate_path_user_id( f: Callable[Concatenate[Request, _P], Coroutine[Any, Any, _T]], ) -> Callable[Concatenate[Request, _P], Coroutine[Any, Any, _T]]: diff --git a/components/renku_data_services/project/blueprints.py b/components/renku_data_services/project/blueprints.py index e07a82ffc..e45e1c4df 100644 --- a/components/renku_data_services/project/blueprints.py +++ b/components/renku_data_services/project/blueprints.py @@ -13,7 +13,6 @@ from renku_data_services.base_api.auth import ( authenticate, only_authenticated, - validate_path_project_id, validate_path_user_id, ) from renku_data_services.base_api.blueprint import BlueprintFactoryResponse, CustomBlueprint @@ -80,12 +79,11 @@ def get_one(self) -> BlueprintFactoryResponse: """Get a specific project.""" @authenticate(self.authenticator) - @validate_path_project_id @extract_if_none_match async def _get_one( - _: Request, user: base_models.APIUser, project_id: str, etag: str | None + _: Request, user: base_models.APIUser, project_id: ULID, etag: str | None ) -> JSONResponse | HTTPResponse: - project = await self.project_repo.get_project(user=user, project_id=ULID.from_str(project_id)) + project = await self.project_repo.get_project(user=user, project_id=project_id) if project.etag is not None and project.etag == etag: return HTTPResponse(status=304) @@ -210,14 +208,11 @@ def get_permissions(self) -> BlueprintFactoryResponse: """Get the permissions of the current user on the project.""" @authenticate(self.authenticator) - @validate_path_project_id - async def _get_permissions(_: Request, user: base_models.APIUser, project_id: str) -> JSONResponse: - permissions = await self.project_repo.get_project_permissions( - user=user, project_id=ULID.from_str(project_id) - ) + async def _get_permissions(_: Request, user: base_models.APIUser, project_id: ULID) -> JSONResponse: + permissions = await self.project_repo.get_project_permissions(user=user, project_id=project_id) return validated_json(apispec.ProjectPermissions, permissions) - return "/projects//permissions", ["GET"], _get_permissions + return "/projects//permissions", ["GET"], _get_permissions @staticmethod def _dump_project(project: project_models.Project) -> dict[str, Any]: