.gitlab-ci.yml

@@ -62,7 +62,7 @@ integration-test-job: # Runs integration tests with the database
         MYSQL_DATABASE: "$CLOWM_DB__NAME"
         MYSQL_USER: "$CLOWM_DB__USER"
         MYSQL_PASSWORD: "$CLOWM_DB__PASSWORD"
-    - name: $CI_REGISTRY/cmg/clowm/clowm-database:v3.1
+    - name: $CI_REGISTRY/cmg/clowm/clowm-database:v3.2
       alias: upgrade-db
       variables:
         DB_HOST: "$CLOWM_DB__HOST"
@@ -95,7 +95,7 @@ e2e-test-job: # Runs e2e tests on the API endpoints
         MYSQL_DATABASE: "$CLOWM_DB__NAME"
         MYSQL_USER: "$CLOWM_DB__USER"
         MYSQL_PASSWORD: "$CLOWM_DB__PASSWORD"
-    - name: $CI_REGISTRY/cmg/clowm/clowm-database:v3.1
+    - name: $CI_REGISTRY/cmg/clowm/clowm-database:v3.2
       alias: upgrade-db
       variables:
         DB_HOST: "$CLOWM_DB__HOST"

Dockerfile

@@ -6,7 +6,7 @@ EXPOSE $PORT
 RUN apt-get update && apt-get -y install dumb-init && apt-get clean
 ENTRYPOINT ["/usr/bin/dumb-init", "--"]
 STOPSIGNAL SIGINT
-RUN pip install --disable-pip-version-check --no-cache-dir httpx[cli] "uvicorn<0.30.0"
+RUN pip install --disable-pip-version-check --no-cache-dir httpx[cli] "uvicorn[standard]<0.30.0"
 
 HEALTHCHECK --interval=5s --timeout=2s CMD httpx http://localhost:$PORT/health || exit 1
 

Dockerfile-Gunicorn

@@ -4,7 +4,7 @@ EXPOSE $PORT
 WORKDIR /app/
 ENV PYTHONPATH=/app
 
-RUN pip install --disable-pip-version-check --no-cache-dir httpx[cli] "gunicorn<21.3.0" "uvicorn<0.30.0"
+RUN pip install --disable-pip-version-check --no-cache-dir httpx[cli] "gunicorn<21.3.0" "uvicorn[standard]<0.30.0"
 COPY ./gunicorn_conf.py /app/gunicorn_conf.py
 COPY ./start_service_gunicorn.sh /app/entrypoint.sh
 

README.md

@@ -3,9 +3,8 @@
 ## Description
 
 Openstack is shipping with an integrated UI to access the Object Store provided by Ceph. Unfortunately, this UI does not
-allow
-fine-grained control who can access a bucket or object. You can either make it accessible for everyone or nobody, but
-Ceph can do this and much more. 👎
+allow fine-grained control who can access a bucket or object. You can either make it accessible for everyone or nobody,
+but Ceph can do this and much more. 👎
 This is the backend for a new UI which can leverage the additional powerful functionality provided by Ceph in a
 user-friendly manner. 👍
 
@@ -51,8 +50,8 @@ user-friendly manner. 👍
 | * `CLOWM_S3__ACCESS_KEY`       | `s3.acess_key`        | unset   | String   | `ZR7U56KMK20VW`          | Access key for the S3 that owns the buckets                                      |
 | * `CLOWM_S3__SECRET_KEY`       | `s3.secret_key`       | unset   | String   | `9KRUU41EGSCB3H9ODECNHW` | Secret key for the S3 that owns the buckets                                      |
 | * `CLOWM_S3__USERNAME`         | `s3.username`         | unset   | String   | `clowm-bucket-manager`   | ID of the user in ceph who owns all the buckets. Owner of `CLOWM_S3__ACCESS_KEY` |
-| * `CLOWM_S3__ADMIN_ACCESS_KEY` | `s3.admin_acess_key`  | unset   | String   | `ZR7U56KMK20VW`          | Access key for the Ceph Object Gateway user with `user:*` privileges             |
-| * `CLOWM_S3__ADMIN_SECRET_KEY` | `s3.admin_secret_key` | unset   | String   | `9KRUU41EGSCB3H9ODECNHW` | Secret key for the Ceph Object Gateway user with `user:*` privileges.            |
+| * `CLOWM_S3__ADMIN_ACCESS_KEY` | `s3.admin_acess_key`  | unset   | String   | `ZR7U56KMK20VW`          | Access key for the Ceph Object Gateway user with `user=*,bucket=*` capabilities. |
+| * `CLOWM_S3__ADMIN_SECRET_KEY` | `s3.admin_secret_key` | unset   | String   | `9KRUU41EGSCB3H9ODECNHW` | Secret key for the Ceph Object Gateway user with `user=*,bucket=*` capabilities. |
 
 ### Security
 
@@ -68,7 +67,6 @@ user-friendly manner. 👍
 | `CLOWM_OTLP__GRPC_ENDPOINT` | `otlp.grpc_endpoint` | unset   | String  | `localhost` | OTLP compatible endpoint to send traces via gRPC, e.g. Jaeger. If unset, no traces are sent. |
 | `CLOWM_OTLP__SECURE`        | `otlp.secure`        | `false` | Boolean | `false`     | Connection type                                                                              |
 
-
 ## License
 
 The API is licensed under the [Apache 2.0](https://www.apache.org/licenses/LICENSE-2.0) license. See

app/api/dependencies.py

@@ -145,7 +145,7 @@ async def get_current_user(token: Annotated[JWT, Depends(decode_bearer_token)],
         uid = UUID(token.sub)
     except ValueError:  # pragma: no cover
         raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="Malformed JWT")
-    user = await CRUDUser.get(db, uid)
+    user = await CRUDUser.get(uid, db=db)
     if user:
         return user
     raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="User not found")
@@ -219,7 +219,7 @@ async def get_user_by_path_uid(
         User with the given uid.
 
     """
-    user = await CRUDUser.get(db, uid)
+    user = await CRUDUser.get(uid, db=db)
     if user:
         return user
     raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="User not found")
@@ -252,7 +252,7 @@ async def get_current_bucket(
     bucket : clowmdb.models.Bucket
         Bucket with the given name.
     """
-    bucket = await CRUDBucket.get(db, bucket_name)
+    bucket = await CRUDBucket.get(bucket_name, db=db)
     if bucket is None:
         raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Bucket not found")
     return bucket

app/api/endpoints/bucket_permissions.py

@@ -17,8 +17,7 @@ from app.api.dependencies import (
     get_user_by_path_uid,
 )
 from app.ceph.s3 import get_s3_bucket_policy, put_s3_bucket_policy
-from app.crud import DuplicateError
-from app.crud.crud_bucket_permission import CRUDBucketPermission
+from app.crud import CRUDBucketPermission, DuplicateError
 from app.otlp import start_as_current_span_async
 from app.schemas.bucket_permission import BucketPermissionIn, BucketPermissionOut, BucketPermissionParameters
 
@@ -75,7 +74,7 @@ async def list_permissions(
     rbac_operation = "list_all"
     await authorization(rbac_operation)
     bucket_permissions = await CRUDBucketPermission.list(
-        db, permission_types=permission_types, permission_status=permission_status
+        db=db, permission_types=permission_types, permission_status=permission_status
     )
     return [BucketPermissionOut.from_db_model(p) for p in bucket_permissions]
 
@@ -123,11 +122,9 @@ async def create_permission(
     target_bucket = await get_current_bucket(permission.bucket_name, db=db)  # Check if the target bucket exists
     if target_bucket.owner_id != current_user.uid:
         raise HTTPException(status.HTTP_403_FORBIDDEN, detail="Action forbidden.")
-    if target_bucket.owner_constraint is not None:
-        raise HTTPException(status.HTTP_403_FORBIDDEN, detail="Initial Buckets can be target of Bucket Permissions.")
     await get_user_by_path_uid(permission.uid, db)  # Check if target user exists
     try:
-        permission_db = await CRUDBucketPermission.create(db, permission)
+        permission_db = await CRUDBucketPermission.create(permission, db=db)
     except ValueError as e:
         current_span.record_exception(e)
         raise HTTPException(
@@ -202,7 +199,7 @@ async def list_permissions_per_user(
     rbac_operation = "list_user" if user == current_user else "list_all"
     await authorization(rbac_operation)
     bucket_permissions = await CRUDBucketPermission.list(
-        db, uid=user.uid, permission_types=permission_types, permission_status=permission_status
+        db=db, uid=user.uid, permission_types=permission_types, permission_status=permission_status
     )
     return [BucketPermissionOut.from_db_model(p) for p in bucket_permissions]
 
@@ -262,7 +259,7 @@ async def list_permissions_per_bucket(
     rbac_operation = "list_bucket" if bucket.owner_id == current_user.uid else "list_all"
     await authorization(rbac_operation)
     bucket_permissions = await CRUDBucketPermission.list(
-        db, bucket_name=bucket.name, permission_types=permission_types, permission_status=permission_status
+        db=db, bucket_name=bucket.name, permission_types=permission_types, permission_status=permission_status
     )
     return [BucketPermissionOut.from_db_model(p) for p in bucket_permissions]
 
@@ -308,7 +305,7 @@ async def get_permission_for_bucket(
     trace.get_current_span().set_attributes({"bucket_name": bucket.name, "uid": str(user.uid)})
     rbac_operation = "read" if user == current_user or current_user.uid == bucket.owner_id else "read_any"
     await authorization(rbac_operation)
-    bucket_permission = await CRUDBucketPermission.get(db, bucket.name, user.uid)
+    bucket_permission = await CRUDBucketPermission.get(bucket.name, user.uid, db=db)
     if bucket_permission:
         return BucketPermissionOut.from_db_model(bucket_permission)
     raise HTTPException(
@@ -360,13 +357,13 @@ async def delete_permission(
     trace.get_current_span().set_attributes({"bucket_name": bucket.name, "uid": str(user.uid)})
     rbac_operation = "delete" if user == current_user or current_user.uid == bucket.owner_id else "delete_any"
     await authorization(rbac_operation)
-    bucket_permission = await CRUDBucketPermission.get(db, bucket.name, user.uid)
+    bucket_permission = await CRUDBucketPermission.get(bucket.name, user.uid, db=db)
     if bucket_permission is None:
         raise HTTPException(
             status.HTTP_404_NOT_FOUND,
             detail=f"Permission for combination of bucket={bucket.name} and user={str(user.uid)} doesn't exists",
         )
-    await CRUDBucketPermission.delete(db, bucket_name=bucket_permission.bucket_name, uid=bucket_permission.uid)
+    await CRUDBucketPermission.delete(db=db, bucket_name=bucket_permission.bucket_name, uid=bucket_permission.uid)
     bucket_permission_schema = BucketPermissionOut.from_db_model(bucket_permission)
     s3_policy = get_s3_bucket_policy(s3, bucket_name=bucket_permission_schema.bucket_name)
     policy = json.loads(s3_policy.policy)
@@ -423,14 +420,14 @@ async def update_permission(
     await authorization("update")
     if bucket.owner_id != current_user.uid:
         raise HTTPException(status.HTTP_403_FORBIDDEN, "Action forbidden")
-    bucket_permission = await CRUDBucketPermission.get(db, bucket.name, user.uid)
+    bucket_permission = await CRUDBucketPermission.get(bucket.name, user.uid, db=db)
 
     if bucket_permission is None:
         raise HTTPException(
             status.HTTP_404_NOT_FOUND,
             detail=f"Permission for combination of bucket={bucket.name} and user={user.uid} doesn't exists",
         )
-    updated_permission = await CRUDBucketPermission.update_permission(db, bucket_permission, permission_parameters)
+    updated_permission = await CRUDBucketPermission.update_permission(bucket_permission, permission_parameters, db=db)
     updated_permission_schema = BucketPermissionOut.from_db_model(updated_permission)
     s3_policy = get_s3_bucket_policy(s3, bucket_name=bucket.name)
     policy = json.loads(s3_policy.policy)

app/api/endpoints/buckets.py

@@ -6,17 +6,24 @@ from botocore.exceptions import ClientError
 from clowmdb.models import Bucket
 from fastapi import APIRouter, Body, Depends, HTTPException, Query, status
 from opentelemetry import trace
+from pydantic import ByteSize
 from pydantic.json_schema import SkipJsonSchema
 
-from app.api.dependencies import AuthorizationDependency, CurrentBucket, CurrentUser, DBSession, S3Resource
+from app.api.dependencies import (
+    AuthorizationDependency,
+    CurrentBucket,
+    CurrentUser,
+    DBSession,
+    RGWAdminResource,
+    S3Resource,
+)
 from app.ceph.s3 import get_s3_bucket_objects, get_s3_bucket_policy, put_s3_bucket_policy
 from app.core.config import settings
-from app.crud import DuplicateError
-from app.crud.crud_bucket import CRUDBucket
-from app.crud.crud_bucket_permission import CRUDBucketPermission
+from app.crud import CRUDBucket, CRUDBucketPermission, DuplicateError
 from app.otlp import start_as_current_span_async
 from app.schemas.bucket import BucketIn as BucketInSchema
 from app.schemas.bucket import BucketOut as BucketOutSchema
+from app.schemas.bucket import BucketSizeLimits
 
 router = APIRouter(prefix="/buckets", tags=["Bucket"])
 bucket_authorization = AuthorizationDependency(resource="bucket")
@@ -116,9 +123,9 @@ async def list_buckets(
     current_span.set_attribute("bucket_type", bucket_type.name)
     await authorization("list_all" if current_user.uid != owner_id else "list")
     if owner_id is None:
-        buckets = await CRUDBucket.get_all(db)
+        buckets = await CRUDBucket.get_all(db=db)
     else:
-        buckets = await CRUDBucket.get_for_user(db, owner_id, bucket_type)
+        buckets = await CRUDBucket.get_for_user(owner_id, bucket_type, db=db)
 
     return buckets
 
@@ -166,7 +173,7 @@ async def create_bucket(
     current_span.set_attribute("bucket_name", bucket.name)
     await authorization("create")
     try:
-        db_bucket = await CRUDBucket.create(db, bucket, current_user.uid)
+        db_bucket = await CRUDBucket.create(bucket, current_user.uid, db=db)
     except DuplicateError as e:
         current_span.record_exception(e)
         raise HTTPException(
@@ -238,14 +245,14 @@ async def get_bucket(
     trace.get_current_span().set_attribute("bucket_name", bucket.name)
     rbac_operation = (
         "read_any"
-        if not bucket.public and not await CRUDBucketPermission.check_permission(db, bucket.name, current_user.uid)
+        if not bucket.public and not await CRUDBucketPermission.check_permission(bucket.name, current_user.uid, db=db)
         else "read"
     )
     await authorization(rbac_operation)
     return bucket
 
 
-@router.patch("/{bucket_name}/public", response_model=BucketOutSchema, summary="update public status")
+@router.patch("/{bucket_name}/public", response_model=BucketOutSchema, summary="Update public status")
 @start_as_current_span_async("api_update_bucket_public_state", tracer=tracer)
 async def update_bucket_public_state(
     bucket: CurrentBucket,
@@ -256,7 +263,7 @@ async def update_bucket_public_state(
     public: Annotated[bool, Body(..., embed=True, description="New State")],
 ) -> Bucket:
     """
-    Toggle the buckets public state. A bucket with an owner constraint can't be made public.\n
+    Update the buckets public state.\n
     Permission `bucket:update` required if the current user is the owner of the bucket,
     otherwise `bucket:update_any` required.
     \f
@@ -283,11 +290,6 @@ async def update_bucket_public_state(
     trace.get_current_span().set_attributes({"bucket_name": bucket.name, "public": public})
     rbac_operation = "update" if bucket.owner_id == current_user.uid else "update_any"
     await authorization(rbac_operation)
-    if bucket.owner_constraint is not None:
-        raise HTTPException(
-            status_code=status.HTTP_400_BAD_REQUEST,
-            detail=f"can't change the public state of a bucket with owner constraint {bucket.owner_constraint}",
-        )
     if bucket.public == public:
         return bucket
     s3_policy = get_s3_bucket_policy(s3, bucket_name=bucket.name)
@@ -302,6 +304,68 @@ async def update_bucket_public_state(
     return bucket
 
 
+@router.patch("/{bucket_name}/limits", response_model=BucketOutSchema, summary="Update bucket limits")
+@start_as_current_span_async("api_update_bucket_limits", tracer=tracer)
+async def update_bucket_limits(
+    bucket: CurrentBucket,
+    authorization: Authorization,
+    db: DBSession,
+    rgw: RGWAdminResource,
+    limits: BucketSizeLimits,
+) -> Bucket:
+    """
+    Update the buckets size limits.\n
+    Permission `bucket:update_any` required.
+    \f
+    Parameters
+    ----------
+    bucket : clowmdb.models.Bucket
+        Bucket with the name provided in the URL path. Dependency Injection.
+    authorization : Callable[[str], Awaitable[Any]]
+        Async function to ask the auth service for authorization. Dependency Injection.
+    db : sqlalchemy.ext.asyncio.AsyncSession.
+        Async database session to perform query on. Dependency Injection.
+    rgw : rgwadmin.RGWAdmin
+        RGW admin interface to manage Ceph's object store. Dependency Injection.
+    limits : app.schemas.bucket.BucketSizeLimits
+        New bucket limits. HTTP Body.
+
+    Returns
+    -------
+    bucket : clowmdb.models.Bucket
+        Bucket with the toggled public state.
+    """
+    current_span = trace.get_current_span()
+    current_span.set_attribute("bucket_name", bucket.name)
+    if limits.size_limit is not None:  # pragma: no cover
+        current_span.set_attribute("size_limit", ByteSize(limits.size_limit * 1024).human_readable())
+    if limits.object_limit is not None:  # pragma: no cover
+        current_span.set_attribute("object_limit", limits.object_limit)
+    await authorization("update_any")
+    with tracer.start_as_current_span(
+        "rgw_set_bucket_limits",
+        attributes={
+            "bucket_name": bucket.name,
+            "enabled": limits.object_limit is not None or limits.size_limit is not None,
+        },
+    ) as span:
+        if limits.size_limit is not None:  # pragma: no cover
+            span.set_attribute("size_limit", ByteSize(limits.size_limit * 1024).human_readable())
+        if limits.object_limit is not None:  # pragma: no cover
+            span.set_attribute("object_limit", limits.object_limit)
+        rgw.set_bucket_quota(
+            uid=str(bucket.owner_id),
+            bucket=bucket.name,
+            max_size_kb=-1 if limits.size_limit is None else limits.size_limit,
+            max_objects=-1 if limits.object_limit is None else limits.object_limit,
+            enabled=limits.object_limit is not None or limits.size_limit is not None,
+        )
+    await CRUDBucket.update_bucket_limits(
+        db=db, bucket_name=bucket.name, object_limit=limits.object_limit, size_limit=limits.size_limit
+    )
+    return bucket
+
+
 @router.delete("/{bucket_name}", status_code=status.HTTP_204_NO_CONTENT, summary="Delete a bucket")
 @start_as_current_span_async("api_delete_bucket", tracer=tracer)
 async def delete_bucket(
@@ -345,6 +409,6 @@ async def delete_bucket(
         with tracer.start_as_current_span("s3_delete_bucket") as span:
             span.set_attribute("bucket_name", bucket.name)
             s3.Bucket(name=bucket.name).delete()
-        await CRUDBucket.delete(db, bucket.name)
+        await CRUDBucket.delete(bucket.name, db=db)
     except ClientError:
         raise HTTPException(status.HTTP_400_BAD_REQUEST, detail="Bucket not empty")

app/core/config.py

@@ -52,10 +52,10 @@ class S3Settings(BaseModel):
         ..., description="ID of the user in ceph who owns all the buckets. Owner of 'CLOWM_S3__ACCESS_KEY'"
     )
     admin_access_key: str = Field(
-        ..., description="Access key for the Ceph Object Gateway with 'user:read,user:write privileges'."
+        ..., description="Access key for the Ceph Object Gateway user with `user=*,bucket=*` capabilities."
     )
     admin_secret_key: SecretStr = Field(
-        ..., description="Secret key for the Ceph Object Gateway with 'user:read,user:write privileges'."
+        ..., description="Secret key for the Ceph Object Gateway user with `user=*,bucket=*` capabilities."
     )
 
 
@@ -91,7 +91,7 @@ class EmailSettings(BaseModel):
 
 
 class Settings(BaseSettings):
-    api_prefix: str = Field("/api/s3proxy-service", description="Path Prefix for all API endpoints.")
+    api_prefix: str = Field("", description="Path Prefix for all API endpoints.")
     public_key: SecretStr | None = Field(None, description="Public RSA Key in PEM format to verify the JWTs.")
     public_key_file: FilePath | None = Field(
         None, description="Path to Public RSA Key in PEM format to verify the JWTs."
@@ -138,6 +138,7 @@ class Settings(BaseSettings):
     ) -> tuple[PydanticBaseSettingsSource, ...]:
         return (
             env_settings,
+            file_secret_settings,
             dotenv_settings,
             YamlConfigSettingsSource(settings_cls),
             TomlConfigSettingsSource(settings_cls),

app/crud/__init__.py

-class DuplicateError(Exception):
-    pass
+from .crud_bucket import CRUDBucket
+from .crud_bucket_permission import CRUDBucketPermission
+from .crud_error import DuplicateError
+from .crud_user import CRUDUser
+
+__all__ = ["CRUDBucketPermission", "CRUDUser", "CRUDBucket", "DuplicateError"]

app/crud/crud_bucket.py

-from enum import Enum, unique
+from enum import StrEnum, unique
 from typing import Sequence
 from uuid import UUID
 
 from clowmdb.models import Bucket
 from clowmdb.models import BucketPermission as BucketPermissionDB
 from opentelemetry import trace
+from pydantic import ByteSize
 from sqlalchemy import delete, func, or_, select, update
 from sqlalchemy.ext.asyncio import AsyncSession
 
-from app.crud import DuplicateError
+from app.crud.crud_error import DuplicateError
 from app.schemas.bucket import BucketIn as BucketInSchema
 
 tracer = trace.get_tracer_provider().get_tracer(__name__)
@@ -16,7 +17,7 @@ tracer = trace.get_tracer_provider().get_tracer(__name__)
 
 class CRUDBucket:
     @unique
-    class BucketType(str, Enum):
+    class BucketType(StrEnum):
         """
         Enumeration for the type of buckets to fetch from the DB
 
@@ -25,12 +26,12 @@ class CRUDBucket:
         ALL: Fetch all buckets that the user has access to
         """
 
-        OWN: str = "OWN"
-        ALL: str = "ALL"
-        PERMISSION: str = "PERMISSION"
+        OWN = "OWN"
+        ALL = "ALL"
+        PERMISSION = "PERMISSION"
 
     @staticmethod
-    async def get(db: AsyncSession, bucket_name: str) -> Bucket | None:
+    async def get(bucket_name: str, *, db: AsyncSession) -> Bucket | None:
         """
         Get a bucket by its name.
 
@@ -59,7 +60,9 @@ class CRUDBucket:
             return (await db.scalars(stmt)).all()
 
     @staticmethod
-    async def get_for_user(db: AsyncSession, uid: UUID, bucket_type: BucketType = BucketType.ALL) -> Sequence[Bucket]:
+    async def get_for_user(
+        uid: UUID, bucket_type: BucketType = BucketType.ALL, *, db: AsyncSession
+    ) -> Sequence[Bucket]:
         """
         Get all buckets for a user. Depending on the `bucket_type`, the user is either owner of the bucket or has
         permission for the bucket
@@ -154,7 +157,7 @@ class CRUDBucket:
             return (await db.scalars(stmt)).all()
 
     @staticmethod
-    async def create(db: AsyncSession, bucket_in: BucketInSchema, uid: UUID) -> Bucket:
+    async def create(bucket_in: BucketInSchema, uid: UUID, *, db: AsyncSession) -> Bucket:
         """
         Create a bucket for a given user.
 
@@ -177,14 +180,14 @@ class CRUDBucket:
             "db_create_bucket",
             attributes={"uid": str(uid), "bucket_name": bucket.name},
         ):
-            if await CRUDBucket.get(db, bucket.name) is not None:
+            if await CRUDBucket.get(bucket.name, db=db) is not None:
                 raise DuplicateError(f"Bucket {bucket.name} exists already")
             db.add(bucket)
             await db.commit()
             return bucket
 
     @staticmethod
-    async def update_public_state(db: AsyncSession, bucket_name: str, public: bool) -> None:
+    async def update_public_state(bucket_name: str, public: bool, *, db: AsyncSession) -> None:
         """
         Update the public state of a bucket
 
@@ -200,13 +203,42 @@ class CRUDBucket:
         stmt = update(Bucket).where(Bucket.name == bucket_name).values(public=public)
         with tracer.start_as_current_span(
             "db_update_bucket_public_state",
-            attributes={"bucket_name": bucket_name, "public": public},
+            attributes={"bucket_name": bucket_name, "public": public, "sql_query": str(stmt)},
         ):
             await db.execute(stmt)
             await db.commit()
 
     @staticmethod
-    async def delete(db: AsyncSession, bucket_name: str) -> None:
+    async def update_bucket_limits(
+        bucket_name: str, size_limit: int | None = None, object_limit: int | None = None, *, db: AsyncSession
+    ) -> None:
+        """
+        Update the bucket limits.
+
+        Parameters
+        ----------
+        db : sqlalchemy.ext.asyncio.AsyncSession
+            Async database session to perform query on.
+        bucket_name : str
+            Name of a bucket.
+        size_limit : int | None, default None
+            New size limit for the bucket.
+        object_limit : int | None, default None
+            New object limit for the bucket.
+        """
+        stmt = update(Bucket).where(Bucket.name == bucket_name).values(size_limit=size_limit, object_limit=object_limit)
+        with tracer.start_as_current_span(
+            "db_update_bucket_limits", attributes={"bucket_name": bucket_name, "sql_query": str(stmt)}
+        ) as span:
+            if size_limit is not None:  # pragma: no cover
+                span.set_attribute("size_limit", ByteSize(size_limit * 1024).human_readable())
+            if object_limit is not None:  # pragma: no cover
+                span.set_attribute("object_limit", object_limit)
+            await db.execute(stmt)
+            await db.commit()
+
+    @staticmethod
+    async def delete(bucket_name: str, *, db: AsyncSession) -> None:
         """
         Delete a specific bucket.
 

app/crud/crud_bucket_permission.py

-from enum import Enum, unique
+from enum import StrEnum, unique
 from typing import Sequence
 from uuid import UUID
 
@@ -9,8 +9,8 @@ from sqlalchemy.ext.asyncio import AsyncSession
 from sqlalchemy.orm import joinedload
 from sqlalchemy.sql import Select as SQLSelect
 
-from app.crud import DuplicateError
 from app.crud.crud_bucket import CRUDBucket
+from app.crud.crud_error import DuplicateError
 from app.crud.crud_user import CRUDUser
 from app.otlp import start_as_current_span_async
 from app.schemas.bucket_permission import BucketPermissionIn as BucketPermissionSchema
@@ -21,17 +21,17 @@ tracer = trace.get_tracer_provider().get_tracer(__name__)
 
 class CRUDBucketPermission:
     @unique
-    class PermissionStatus(str, Enum):
+    class PermissionStatus(StrEnum):
         """
         Status of a bucket permission. Can be either `ACTIVE` or `INACTIVE`. A permission can only get `INACTIVE` if the
         permission itself has a time limit and the current time is not in the timespan.
         """
 
-        ACTIVE: str = "ACTIVE"
-        INACTIVE: str = "INACTIVE"
+        ACTIVE = "ACTIVE"
+        INACTIVE = "INACTIVE"
 
     @staticmethod
-    async def get(db: AsyncSession, bucket_name: str, uid: UUID) -> BucketPermissionDB | None:
+    async def get(bucket_name: str, uid: UUID, *, db: AsyncSession) -> BucketPermissionDB | None:
         stmt = select(BucketPermissionDB).where(
             BucketPermissionDB.uid_bytes == uid.bytes, BucketPermissionDB.bucket_name == bucket_name
         )
@@ -44,11 +44,12 @@ class CRUDBucketPermission:
     @staticmethod
     @start_as_current_span_async("db_list_bucket_permissions", tracer=tracer)
     async def list(
-        db: AsyncSession,
         bucket_name: str | None = None,
         uid: UUID | None = None,
         permission_types: list[BucketPermissionDB.Permission] | None = None,
         permission_status: PermissionStatus | None = None,
+        *,
+        db: AsyncSession,
     ) -> Sequence[BucketPermissionDB]:
         """
         Get the permissions for the given bucket.
@@ -92,7 +93,7 @@ class CRUDBucketPermission:
         return (await db.scalars(stmt)).all()
 
     @staticmethod
-    async def check_permission(db: AsyncSession, bucket_name: str, uid: UUID) -> bool:
+    async def check_permission(bucket_name: str, uid: UUID, *, db: AsyncSession) -> bool:
         """
         Check if the provided user has any permission to the provided bucket.
 
@@ -114,12 +115,12 @@ class CRUDBucketPermission:
             "db_check_bucket_permission",
             attributes={"uid": str(uid), "bucket_name": bucket_name},
         ):
-            buckets = await CRUDBucket.get_for_user(db, uid, bucket_type=CRUDBucket.BucketType.ALL)
+            buckets = await CRUDBucket.get_for_user(uid, bucket_type=CRUDBucket.BucketType.ALL, db=db)
             return bucket_name in map(lambda x: x.name, buckets)
 
     @staticmethod
     @start_as_current_span_async("db_create_bucket_permission", tracer=tracer)
-    async def create(db: AsyncSession, permission: BucketPermissionSchema) -> BucketPermissionDB:
+    async def create(permission: BucketPermissionSchema, *, db: AsyncSession) -> BucketPermissionDB:
         """
         Create a permission in the database and raise Exceptions if there are problems.
 
@@ -136,17 +137,17 @@ class CRUDBucketPermission:
         """
         trace.get_current_span().set_attributes({"bucket_name": permission.bucket_name, "uid": str(permission.uid)})
         # Check if user exists
-        user = await CRUDUser.get(db, uid=permission.uid)
+        user = await CRUDUser.get(uid=permission.uid, db=db)
         if user is None:
             raise KeyError(
                 f"Unknown user with uid {str(permission.uid)}",
             )
         # Check that grantee is not the owner of the bucket
-        bucket = await CRUDBucket.get(db, permission.bucket_name)
+        bucket = await CRUDBucket.get(permission.bucket_name, db=db)
         if bucket is None or bucket.owner_id == user.uid:
             raise ValueError(f"User {str(permission.uid)} is the owner of the bucket {permission.bucket_name}")
         # Check if combination of user and bucket already exists
-        previous_permission = await CRUDBucketPermission.get(db, bucket_name=permission.bucket_name, uid=user.uid)
+        previous_permission = await CRUDBucketPermission.get(bucket_name=permission.bucket_name, uid=user.uid, db=db)
         if previous_permission is not None:
             raise DuplicateError(
                 f"bucket permission for combination {permission.bucket_name} {str(permission.uid)} already exists."
@@ -166,7 +167,7 @@ class CRUDBucketPermission:
         return permission_db
 
     @staticmethod
-    async def delete(db: AsyncSession, bucket_name: str, uid: UUID) -> None:
+    async def delete(bucket_name: str, uid: UUID, *, db: AsyncSession) -> None:
         """
         Delete a permission in the database.
 
@@ -191,7 +192,7 @@ class CRUDBucketPermission:
 
     @staticmethod
     async def update_permission(
-        db: AsyncSession, permission: BucketPermissionDB, new_params: BucketPermissionParametersSchema
+        permission: BucketPermissionDB, new_params: BucketPermissionParametersSchema, *, db: AsyncSession
     ) -> BucketPermissionDB:
         """
         Update a permission in the database.

app/crud/crud_error.py

+class DuplicateError(Exception):
+    pass

app/crud/crud_user.py

@@ -10,7 +10,7 @@ tracer = trace.get_tracer_provider().get_tracer(__name__)
 
 class CRUDUser:
     @staticmethod
-    async def get(db: AsyncSession, uid: UUID) -> User | None:
+    async def get(uid: UUID, *, db: AsyncSession) -> User | None:
         """
         Get a user by its UID.
 

app/schemas/bucket.py

 import re
 
-from clowmdb.models import Bucket
 from pydantic import BaseModel, ConfigDict, Field, field_validator
 
 from app.schemas import UUID
@@ -44,7 +43,16 @@ class BucketIn(_BaseBucket):
     """
 
 
-class BucketOut(_BaseBucket):
+class BucketSizeLimits(BaseModel):
+    size_limit: int | None = Field(
+        None, gt=0, lt=2**32, description="Size limit of the bucket in KiB", examples=[10240]
+    )
+    object_limit: int | None = Field(
+        None, gt=0, lt=2**32, description="Number of objects limit of the bucket", examples=[10000]
+    )
+
+
+class BucketOut(_BaseBucket, BucketSizeLimits):
     """
     Schema for answering a request with a bucket.
     """
@@ -55,6 +63,5 @@ class BucketOut(_BaseBucket):
         description="Time when the bucket was created as UNIX timestamp",
     )
     owner_id: UUID = Field(..., description="UID of the owner", examples=["1d3387f3-95c0-4813-8767-2cad87faeebf"])
-    owner_constraint: Bucket.Constraint | None = Field(None, description="Constraint for the owner of the bucket")
     public: bool = Field(..., description="Flag if the bucket is anonymously readable")
     model_config = ConfigDict(from_attributes=True)

app/tests/api/test_bucket_permissions.py

@@ -6,7 +6,6 @@ from clowmdb.models import Bucket, BucketPermission
 from fastapi import status
 from httpx import AsyncClient
 from pydantic import TypeAdapter
-from sqlalchemy import update
 from sqlalchemy.ext.asyncio import AsyncSession
 
 from app.schemas.bucket_permission import BucketPermissionIn as BucketPermissionSchema
@@ -423,43 +422,6 @@ class TestBucketPermissionRoutesCreate(_TestBucketPermissionRoutes):
         assert created_permission.uid == random_second_user.user.uid
         assert created_permission.bucket_name == random_bucket.name
 
-    @pytest.mark.asyncio
-    async def test_create_bucket_permissions_on_initial_bucket(
-        self,
-        client: AsyncClient,
-        db: AsyncSession,
-        random_user: UserWithAuthHeader,
-        random_second_user: UserWithAuthHeader,
-        random_bucket: Bucket,
-    ) -> None:
-        """
-        Test for creating a bucket permission on an initial READ Bucket.
-
-        Parameters
-        ----------
-        client : httpx.AsyncClient
-            HTTP Client to perform the request on.
-        db : sqlalchemy.ext.asyncio.AsyncSession.
-            Async database session to perform query on.
-        random_user : app.tests.utils.user.UserWithAuthHeader
-            Random user for testing.
-        random_second_user : app.tests.utils.user.UserWithAuthHeader
-            Random second user for testing.
-        random_bucket : clowmdb.models.Bucket
-            Random bucket for testing.
-        """
-        update_stmt = (
-            update(Bucket).where(Bucket.name == random_bucket.name).values(owner_constraint=Bucket.Constraint.READ)
-        )
-        await db.execute(update_stmt)
-        await db.commit()
-        permission = BucketPermissionSchema(bucket_name=random_bucket.name, uid=random_second_user.user.uid)
-
-        response = await client.post(
-            self.base_path, headers=random_user.auth_headers, content=permission.model_dump_json()
-        )
-        assert response.status_code == status.HTTP_403_FORBIDDEN
-
 
 class TestBucketPermissionRoutesDelete(_TestBucketPermissionRoutes):
     @pytest.mark.asyncio

app/tests/api/test_buckets.py

@@ -3,12 +3,11 @@ from clowmdb.models import Bucket, BucketPermission
 from fastapi import status
 from httpx import AsyncClient
 from pydantic import TypeAdapter
-from sqlalchemy import update
 from sqlalchemy.ext.asyncio import AsyncSession
 
 from app.api.endpoints.buckets import ANONYMOUS_ACCESS_SID
-from app.crud.crud_bucket import CRUDBucket
-from app.schemas.bucket import BucketIn, BucketOut
+from app.crud import CRUDBucket
+from app.schemas.bucket import BucketIn, BucketOut, BucketSizeLimits
 from app.tests.mocks.mock_s3_resource import MockS3ServiceResource
 from app.tests.utils.bucket import add_permission_for_bucket, delete_bucket, make_bucket_public
 from app.tests.utils.cleanup import CleanupList
@@ -216,7 +215,7 @@ class TestBucketRoutesCreate(_TestBucketRoutes):
         assert bucket.name == bucket_info.name
         assert bucket.owner_id == random_user.user.uid
 
-        db_bucket = await CRUDBucket.get(db, bucket_info.name)
+        db_bucket = await CRUDBucket.get(bucket_info.name, db=db)
         assert db_bucket
         assert db_bucket.name == bucket_info.name
         assert db_bucket.owner_id == random_user.user.uid
@@ -284,13 +283,11 @@ class TestBucketRoutesUpdate(_TestBucketRoutes):
         assert ANONYMOUS_ACCESS_SID in mock_s3_service.Bucket(bucket.name).Policy().policy
 
     @pytest.mark.asyncio
-    async def test_make_bucket_private(
+    async def test_update_bucket_limits(
         self,
         client: AsyncClient,
         random_bucket: Bucket,
         random_user: UserWithAuthHeader,
-        mock_s3_service: MockS3ServiceResource,
-        db: AsyncSession,
     ) -> None:
         """
         Test for getting a foreign public bucket.
@@ -303,26 +300,25 @@ class TestBucketRoutesUpdate(_TestBucketRoutes):
             Random bucket for testing.
         random_user : app.tests.utils.user.UserWithAuthHeader
             Random user who is owner of the bucket.
-        mock_s3_service : app.tests.mocks.mock_s3_resource.MockS3ServiceResource
-            Mock S3 Service to manipulate objects.
-        db : sqlalchemy.ext.asyncio.AsyncSession.
-            Async database session to perform query on.
         """
-        await make_bucket_public(db=db, s3=mock_s3_service, bucket_name=random_bucket.name)
         response = await client.patch(
-            f"{self.base_path}/{random_bucket.name}/public", headers=random_user.auth_headers, json={"public": False}
+            f"{self.base_path}/{random_bucket.name}/limits",
+            headers=random_user.auth_headers,
+            content=BucketSizeLimits(size_limit=10240, object_limit=1000).model_dump_json(),
         )
         assert response.status_code == status.HTTP_200_OK
 
         bucket = BucketOut.model_validate_json(response.content)
 
         assert bucket.name == random_bucket.name
-        assert not bucket.public
+        assert bucket.size_limit is not None
+        assert bucket.size_limit == 10240
 
-        assert ANONYMOUS_ACCESS_SID not in mock_s3_service.Bucket(bucket.name).Policy().policy
+        assert bucket.object_limit is not None
+        assert bucket.object_limit == 1000
 
     @pytest.mark.asyncio
-    async def test_make_private_bucket_private(
+    async def test_make_bucket_private(
         self,
         client: AsyncClient,
         random_bucket: Bucket,
@@ -346,6 +342,7 @@ class TestBucketRoutesUpdate(_TestBucketRoutes):
         db : sqlalchemy.ext.asyncio.AsyncSession.
             Async database session to perform query on.
         """
+        await make_bucket_public(db=db, s3=mock_s3_service, bucket_name=random_bucket.name)
         response = await client.patch(
             f"{self.base_path}/{random_bucket.name}/public", headers=random_user.auth_headers, json={"public": False}
         )
@@ -359,11 +356,12 @@ class TestBucketRoutesUpdate(_TestBucketRoutes):
         assert ANONYMOUS_ACCESS_SID not in mock_s3_service.Bucket(bucket.name).Policy().policy
 
     @pytest.mark.asyncio
-    async def test_make_bucket_public_with_owner_constraint(
+    async def test_make_private_bucket_private(
         self,
         client: AsyncClient,
         random_bucket: Bucket,
         random_user: UserWithAuthHeader,
+        mock_s3_service: MockS3ServiceResource,
         db: AsyncSession,
     ) -> None:
         """
@@ -377,19 +375,22 @@ class TestBucketRoutesUpdate(_TestBucketRoutes):
             Random bucket for testing.
         random_user : app.tests.utils.user.UserWithAuthHeader
             Random user who is owner of the bucket.
+        mock_s3_service : app.tests.mocks.mock_s3_resource.MockS3ServiceResource
+            Mock S3 Service to manipulate objects.
         db : sqlalchemy.ext.asyncio.AsyncSession.
             Async database session to perform query on.
         """
-        update_stmt = (
-            update(Bucket).where(Bucket.name == random_bucket.name).values(owner_constraint=Bucket.Constraint.READ)
-        )
-        await db.execute(update_stmt)
-        await db.commit()
-
         response = await client.patch(
-            f"{self.base_path}/{random_bucket.name}/public", headers=random_user.auth_headers, json={"public": True}
+            f"{self.base_path}/{random_bucket.name}/public", headers=random_user.auth_headers, json={"public": False}
         )
-        assert response.status_code == status.HTTP_400_BAD_REQUEST
+        assert response.status_code == status.HTTP_200_OK
+
+        bucket = BucketOut.model_validate_json(response.content)
+
+        assert bucket.name == random_bucket.name
+        assert not bucket.public
+
+        assert ANONYMOUS_ACCESS_SID not in mock_s3_service.Bucket(bucket.name).Policy().policy
 
 
 class TestBucketRoutesDelete(_TestBucketRoutes):

app/tests/crud/test_bucket.py

@@ -5,8 +5,7 @@ from clowmdb.models import Bucket, BucketPermission
 from sqlalchemy import select
 from sqlalchemy.ext.asyncio import AsyncSession
 
-from app.crud import DuplicateError
-from app.crud.crud_bucket import CRUDBucket
+from app.crud import CRUDBucket, DuplicateError
 from app.schemas.bucket import BucketIn
 from app.tests.utils.bucket import add_permission_for_bucket, delete_bucket
 from app.tests.utils.cleanup import CleanupList
@@ -27,7 +26,7 @@ class TestBucketCRUDGet:
         random_bucket : clowmdb.models.Bucket
             Random bucket for testing.
         """
-        buckets = await CRUDBucket.get_all(db)
+        buckets = await CRUDBucket.get_all(db=db)
         assert len(buckets) == 1
         bucket = buckets[0]
         assert bucket.name == random_bucket.name
@@ -46,7 +45,7 @@ class TestBucketCRUDGet:
         random_bucket : clowmdb.models.Bucket
             Random bucket for testing.
         """
-        bucket = await CRUDBucket.get(db, random_bucket.name)
+        bucket = await CRUDBucket.get(random_bucket.name, db=db)
         assert bucket
         assert bucket.name == random_bucket.name
         assert bucket.public == random_bucket.public
@@ -62,7 +61,7 @@ class TestBucketCRUDGet:
         db : sqlalchemy.ext.asyncio.AsyncSession.
             Async database session to perform query on.
         """
-        bucket = await CRUDBucket.get(db, "unknown Bucket")
+        bucket = await CRUDBucket.get("unknown Bucket", db=db)
         assert bucket is None
 
     @pytest.mark.asyncio
@@ -77,7 +76,7 @@ class TestBucketCRUDGet:
         random_bucket : clowmdb.models.Bucket
             Random bucket for testing.
         """
-        buckets = await CRUDBucket.get_for_user(db, random_bucket.owner_id, CRUDBucket.BucketType.OWN)
+        buckets = await CRUDBucket.get_for_user(random_bucket.owner_id, CRUDBucket.BucketType.OWN, db=db)
 
         assert len(buckets) == 1
         assert buckets[0].name == random_bucket.name
@@ -120,7 +119,7 @@ class TestBucketCRUDGet:
             db, bucket.name, random_bucket.owner_id, permission=BucketPermission.Permission.READ
         )
 
-        buckets = await CRUDBucket.get_for_user(db, random_bucket.owner_id, CRUDBucket.BucketType.PERMISSION)
+        buckets = await CRUDBucket.get_for_user(random_bucket.owner_id, CRUDBucket.BucketType.PERMISSION, db=db)
         assert len(buckets) == 1
         assert buckets[0] != random_bucket
         assert buckets[0].name == bucket.name
@@ -163,7 +162,7 @@ class TestBucketCRUDGet:
             db, bucket.name, random_bucket.owner_id, permission=BucketPermission.Permission.READ
         )
 
-        buckets = await CRUDBucket.get_for_user(db, random_bucket.owner_id)
+        buckets = await CRUDBucket.get_for_user(random_bucket.owner_id, db=db)
 
         assert len(buckets) == 2
         assert buckets[0].name == random_bucket.name or buckets[1].name == random_bucket.name
@@ -189,7 +188,7 @@ class TestBucketCRUDGet:
             db, random_bucket.name, random_second_user.user.uid, permission=BucketPermission.Permission.READ
         )
 
-        buckets = await CRUDBucket.get_for_user(db, random_second_user.user.uid)
+        buckets = await CRUDBucket.get_for_user(random_second_user.user.uid, db=db)
 
         assert len(buckets) > 0
         assert buckets[0].name == random_bucket.name
@@ -214,7 +213,7 @@ class TestBucketCRUDGet:
             db, random_bucket.name, random_second_user.user.uid, permission=BucketPermission.Permission.READWRITE
         )
 
-        buckets = await CRUDBucket.get_for_user(db, random_second_user.user.uid)
+        buckets = await CRUDBucket.get_for_user(random_second_user.user.uid, db=db)
 
         assert len(buckets) > 0
         assert buckets[0].name == random_bucket.name
@@ -239,7 +238,7 @@ class TestBucketCRUDGet:
             db, random_bucket.name, random_second_user.user.uid, permission=BucketPermission.Permission.WRITE
         )
 
-        buckets = await CRUDBucket.get_for_user(db, random_second_user.user.uid)
+        buckets = await CRUDBucket.get_for_user(random_second_user.user.uid, db=db)
 
         assert len(buckets) == 1
         assert buckets[0] == random_bucket
@@ -268,7 +267,7 @@ class TestBucketCRUDGet:
             to=datetime.now() + timedelta(days=10),
         )
 
-        buckets = await CRUDBucket.get_for_user(db, random_second_user.user.uid)
+        buckets = await CRUDBucket.get_for_user(random_second_user.user.uid, db=db)
 
         assert len(buckets) > 0
         assert buckets[0].name == random_bucket.name
@@ -293,7 +292,7 @@ class TestBucketCRUDGet:
             db, random_bucket.name, random_second_user.user.uid, from_=datetime.now() + timedelta(days=10)
         )
 
-        buckets = await CRUDBucket.get_for_user(db, random_second_user.user.uid)
+        buckets = await CRUDBucket.get_for_user(random_second_user.user.uid, db=db)
 
         assert len(buckets) == 0
 
@@ -317,7 +316,7 @@ class TestBucketCRUDGet:
             db, random_bucket.name, random_second_user.user.uid, to=datetime.now() - timedelta(days=10)
         )
 
-        buckets = await CRUDBucket.get_for_user(db, random_second_user.user.uid)
+        buckets = await CRUDBucket.get_for_user(random_second_user.user.uid, db=db)
 
         assert len(buckets) == 0
 
@@ -343,7 +342,7 @@ class TestBucketCRUDCreate:
             Cleanup object where (async) functions can be registered which get executed after a (failed) test.
         """
         bucket_info = BucketIn(name=random_lower_string(), description=random_lower_string(127))
-        bucket = await CRUDBucket.create(db, bucket_info, random_user.user.uid)
+        bucket = await CRUDBucket.create(bucket_info, random_user.user.uid, db=db)
         cleanup.add_task(
             delete_bucket,
             db=db,
@@ -375,14 +374,14 @@ class TestBucketCRUDCreate:
         """
         bucket_info = BucketIn(name=random_bucket.name, description=random_lower_string(127))
         with pytest.raises(DuplicateError):
-            await CRUDBucket.create(db, bucket_info, random_bucket.owner_id)
+            await CRUDBucket.create(bucket_info, random_bucket.owner_id, db=db)
 
 
 class TestBucketCRUDUpdate:
     @pytest.mark.asyncio
     async def test_update_public_state(self, db: AsyncSession, random_bucket: Bucket) -> None:
         """
-        Test for deleting a bucket with the CRUD Repository.
+        Test for updating the bucket public state with the CRUD Repository.
 
         Parameters
         ----------
@@ -401,6 +400,31 @@ class TestBucketCRUDUpdate:
         assert bucket_db == random_bucket
         assert old_public_state != bucket_db.public
 
+    @pytest.mark.asyncio
+    async def test_update_bucket_limits(self, db: AsyncSession, random_bucket: Bucket) -> None:
+        """
+        Test for updating the bucket limits with the CRUD Repository.
+
+        Parameters
+        ----------
+        db : sqlalchemy.ext.asyncio.AsyncSession.
+            Async database session to perform query on.
+        random_bucket : clowmdb.models.Bucket
+            Random bucket for testing.
+        """
+        await CRUDBucket.update_bucket_limits(db=db, bucket_name=random_bucket.name, size_limit=100, object_limit=120)
+
+        stmt = select(Bucket).where(Bucket.name == random_bucket.name)
+        bucket_db = await db.scalar(stmt)
+
+        assert bucket_db is not None
+        assert bucket_db == random_bucket
+        assert bucket_db.size_limit is not None
+        assert bucket_db.size_limit == 100
+
+        assert bucket_db.object_limit is not None
+        assert bucket_db.object_limit == 120
+
 
 class TestBucketCRUDDelete:
     @pytest.mark.asyncio
@@ -415,7 +439,7 @@ class TestBucketCRUDDelete:
         random_bucket : clowmdb.models.Bucket
             Random bucket for testing.
         """
-        await CRUDBucket.delete(db, random_bucket.name)
+        await CRUDBucket.delete(random_bucket.name, db=db)
 
         stmt = select(Bucket).where(Bucket.name == random_bucket.name)
         bucket_db = await db.scalar(stmt)

app/tests/crud/test_bucket_permission.py

@@ -7,8 +7,7 @@ from clowmdb.models import BucketPermission as BucketPermissionDB
 from sqlalchemy import and_, select
 from sqlalchemy.ext.asyncio import AsyncSession
 
-from app.crud import DuplicateError
-from app.crud.crud_bucket_permission import CRUDBucketPermission
+from app.crud import CRUDBucketPermission, DuplicateError
 from app.schemas.bucket_permission import BucketPermissionIn as BucketPermissionSchema
 from app.schemas.bucket_permission import BucketPermissionParameters as BucketPermissionParametersSchema
 from app.tests.utils.bucket import add_permission_for_bucket
@@ -31,7 +30,7 @@ class TestBucketPermissionCRUDGet:
             Bucket permission for a random bucket for testing.
         """
         bucket_permission = await CRUDBucketPermission.get(
-            db, bucket_name=random_bucket_permission.bucket_name, uid=random_bucket_permission.uid
+            bucket_name=random_bucket_permission.bucket_name, uid=random_bucket_permission.uid, db=db
         )
         assert bucket_permission
         assert bucket_permission.uid == random_bucket_permission.uid
@@ -52,7 +51,7 @@ class TestBucketPermissionCRUDGet:
         random_bucket_permission : clowmdb.models.BucketPermission
             Bucket permission for a random bucket for testing.
         """
-        bucket_permissions = await CRUDBucketPermission.list(db, bucket_name=random_bucket_permission.bucket_name)
+        bucket_permissions = await CRUDBucketPermission.list(db=db, bucket_name=random_bucket_permission.bucket_name)
         assert len(bucket_permissions) == 1
         bucket_permission = bucket_permissions[0]
         assert bucket_permission.uid == random_bucket_permission.uid
@@ -84,7 +83,7 @@ class TestBucketPermissionCRUDGet:
             db, random_bucket.name, random_third_user.user.uid, permission=BucketPermissionDB.Permission.WRITE
         )
         bucket_permissions = await CRUDBucketPermission.list(
-            db, bucket_name=random_bucket.name, permission_types=[BucketPermissionDB.Permission.READ]
+            db=db, bucket_name=random_bucket.name, permission_types=[BucketPermissionDB.Permission.READ]
         )
         assert len(bucket_permissions) == 1
         assert bucket_permissions[0].uid == random_second_user.user.uid
@@ -114,7 +113,7 @@ class TestBucketPermissionCRUDGet:
             db, random_bucket.name, random_third_user.user.uid, permission=BucketPermissionDB.Permission.WRITE
         )
         bucket_permissions = await CRUDBucketPermission.list(
-            db,
+            db=db,
             bucket_name=random_bucket.name,
             permission_types=[BucketPermissionDB.Permission.READ, BucketPermissionDB.Permission.WRITE],
         )
@@ -150,7 +149,7 @@ class TestBucketPermissionCRUDGet:
             db, random_bucket.name, random_third_user.user.uid, from_=datetime.now() - timedelta(weeks=1)
         )
         bucket_permissions = await CRUDBucketPermission.list(
-            db, bucket_name=random_bucket.name, permission_status=CRUDBucketPermission.PermissionStatus.ACTIVE
+            db=db, bucket_name=random_bucket.name, permission_status=CRUDBucketPermission.PermissionStatus.ACTIVE
         )
         assert len(bucket_permissions) == 1
         assert bucket_permissions[0].uid == random_third_user.user.uid
@@ -183,7 +182,7 @@ class TestBucketPermissionCRUDGet:
             db, random_bucket.name, random_third_user.user.uid, to=datetime.now() + timedelta(weeks=1)
         )
         bucket_permissions = await CRUDBucketPermission.list(
-            db, bucket_name=random_bucket.name, permission_status=CRUDBucketPermission.PermissionStatus.ACTIVE
+            db=db, bucket_name=random_bucket.name, permission_status=CRUDBucketPermission.PermissionStatus.ACTIVE
         )
         assert len(bucket_permissions) == 1
         assert bucket_permissions[0].uid == random_third_user.user.uid
@@ -224,7 +223,7 @@ class TestBucketPermissionCRUDGet:
             from_=datetime.now() - timedelta(weeks=1),
         )
         bucket_permissions = await CRUDBucketPermission.list(
-            db, bucket_name=random_bucket.name, permission_status=CRUDBucketPermission.PermissionStatus.ACTIVE
+            db=db, bucket_name=random_bucket.name, permission_status=CRUDBucketPermission.PermissionStatus.ACTIVE
         )
         assert len(bucket_permissions) == 1
         assert bucket_permissions[0].uid == random_third_user.user.uid
@@ -257,7 +256,7 @@ class TestBucketPermissionCRUDGet:
             db, random_bucket.name, random_third_user.user.uid, from_=datetime.now() - timedelta(weeks=1)
         )
         bucket_permissions = await CRUDBucketPermission.list(
-            db, bucket_name=random_bucket.name, permission_status=CRUDBucketPermission.PermissionStatus.INACTIVE
+            db=db, bucket_name=random_bucket.name, permission_status=CRUDBucketPermission.PermissionStatus.INACTIVE
         )
         assert len(bucket_permissions) == 1
         assert bucket_permissions[0].uid == random_second_user.user.uid
@@ -290,7 +289,7 @@ class TestBucketPermissionCRUDGet:
             db, random_bucket.name, random_third_user.user.uid, to=datetime.now() + timedelta(weeks=1)
         )
         bucket_permissions = await CRUDBucketPermission.list(
-            db, bucket_name=random_bucket.name, permission_status=CRUDBucketPermission.PermissionStatus.INACTIVE
+            db=db, bucket_name=random_bucket.name, permission_status=CRUDBucketPermission.PermissionStatus.INACTIVE
         )
         assert len(bucket_permissions) == 1
         assert bucket_permissions[0].uid == random_second_user.user.uid
@@ -331,7 +330,7 @@ class TestBucketPermissionCRUDGet:
             from_=datetime.now() - timedelta(weeks=1),
         )
         bucket_permissions = await CRUDBucketPermission.list(
-            db, bucket_name=random_bucket.name, permission_status=CRUDBucketPermission.PermissionStatus.INACTIVE
+            db=db, bucket_name=random_bucket.name, permission_status=CRUDBucketPermission.PermissionStatus.INACTIVE
         )
         assert len(bucket_permissions) == 1
         assert bucket_permissions[0].uid == random_second_user.user.uid
@@ -350,7 +349,7 @@ class TestBucketPermissionCRUDGet:
         random_bucket_permission : clowmdb.models.BucketPermission
             Bucket permission for a random bucket for testing.
         """
-        bucket_permissions = await CRUDBucketPermission.list(db, uid=random_bucket_permission.uid)
+        bucket_permissions = await CRUDBucketPermission.list(db=db, uid=random_bucket_permission.uid)
         assert len(bucket_permissions) == 1
         bucket_permission = bucket_permissions[0]
         assert bucket_permission.uid == random_bucket_permission.uid
@@ -378,7 +377,7 @@ class TestBucketPermissionCRUDGet:
             db, random_bucket.name, random_second_user.user.uid, permission=BucketPermissionDB.Permission.WRITE
         )
         bucket_permissions = await CRUDBucketPermission.list(
-            db, uid=random_second_user.user.uid, permission_types=[BucketPermissionDB.Permission.READ]
+            db=db, uid=random_second_user.user.uid, permission_types=[BucketPermissionDB.Permission.READ]
         )
         assert len(bucket_permissions) == 0
 
@@ -405,7 +404,7 @@ class TestBucketPermissionCRUDGet:
         )
 
         bucket_permissions = await CRUDBucketPermission.list(
-            db, uid=random_second_user.user.uid, permission_status=CRUDBucketPermission.PermissionStatus.ACTIVE
+            db=db, uid=random_second_user.user.uid, permission_status=CRUDBucketPermission.PermissionStatus.ACTIVE
         )
         assert len(bucket_permissions) == 0
 
@@ -425,7 +424,7 @@ class TestBucketPermissionCRUDCreate:
         """
         permission = BucketPermissionSchema(bucket_name=random_bucket.name, uid=uuid4())
         with pytest.raises(KeyError):
-            await CRUDBucketPermission.create(db, permission)
+            await CRUDBucketPermission.create(permission, db=db)
 
     @pytest.mark.asyncio
     async def test_create_bucket_permissions_for_owner(
@@ -445,7 +444,7 @@ class TestBucketPermissionCRUDCreate:
         """
         permission = BucketPermissionSchema(bucket_name=random_bucket.name, uid=random_user.user.uid)
         with pytest.raises(ValueError):
-            await CRUDBucketPermission.create(db, permission)
+            await CRUDBucketPermission.create(permission, db=db)
 
     @pytest.mark.asyncio
     async def test_create_duplicate_bucket_permissions(
@@ -465,7 +464,7 @@ class TestBucketPermissionCRUDCreate:
             bucket_name=random_bucket_permission_schema.bucket_name, uid=random_bucket_permission_schema.uid
         )
         with pytest.raises(DuplicateError):
-            await CRUDBucketPermission.create(db, permission)
+            await CRUDBucketPermission.create(permission, db=db)
 
     @pytest.mark.asyncio
     async def test_create_valid_bucket_permissions(
@@ -484,7 +483,7 @@ class TestBucketPermissionCRUDCreate:
             Random bucket for testing.
         """
         permission = BucketPermissionSchema(bucket_name=random_bucket.name, uid=random_second_user.user.uid)
-        created_permission = await CRUDBucketPermission.create(db, permission)
+        created_permission = await CRUDBucketPermission.create(permission, db=db)
 
         assert created_permission.uid == random_second_user.user.uid
         assert created_permission.bucket_name == random_bucket.name
@@ -506,7 +505,7 @@ class TestBucketPermissionCRUDDelete:
             Bucket permission for a random bucket for testing.
         """
         await CRUDBucketPermission.delete(
-            db, bucket_name=random_bucket_permission.bucket_name, uid=random_bucket_permission.uid
+            db=db, bucket_name=random_bucket_permission.bucket_name, uid=random_bucket_permission.uid
         )
 
         stmt = select(BucketPermissionDB).where(
@@ -542,7 +541,7 @@ class TestBucketPermissionCRUDUpdate:
             permission=BucketPermissionDB.Permission.READWRITE,
             file_prefix="pseudo/folder/",
         )
-        new_permission = await CRUDBucketPermission.update_permission(db, random_bucket_permission, new_params)
+        new_permission = await CRUDBucketPermission.update_permission(random_bucket_permission, new_params, db=db)
 
         assert new_permission.uid == random_bucket_permission.uid
         assert new_permission.bucket_name == random_bucket_permission.bucket_name

app/tests/crud/test_user.py

@@ -3,7 +3,7 @@ from uuid import uuid4
 import pytest
 from sqlalchemy.ext.asyncio import AsyncSession
 
-from app.crud.crud_user import CRUDUser
+from app.crud import CRUDUser
 from app.tests.utils.user import UserWithAuthHeader
 
 
@@ -20,7 +20,7 @@ class TestUserCRUD:
         random_user : clowmdb.models.User
             Random user for testing.
         """
-        user = await CRUDUser.get(db, random_user.user.uid)
+        user = await CRUDUser.get(random_user.user.uid, db=db)
         assert user
         assert random_user.user.uid == user.uid
         assert random_user.user.display_name == user.display_name
@@ -38,5 +38,5 @@ class TestUserCRUD:
         db : sqlalchemy.ext.asyncio.AsyncSession.
             Async database session to perform query on.
         """
-        user = await CRUDUser.get(db, uuid4())
+        user = await CRUDUser.get(uuid4(), db=db)
         assert user is None

app/tests/mocks/mock_rgw_admin.py

@@ -25,6 +25,16 @@ class MockRGWAdmin:
     def __init__(self) -> None:
         self._keys = {}
 
+    def set_bucket_quota(
+        self,
+        uid: str,
+        bucket: str,
+        max_size_kb: int | None = None,
+        max_objects: int | None = None,
+        enabled: bool = True,
+    ) -> None:
+        pass
+
     def create_user(self, uid: str, max_buckets: int, display_name: str) -> None:
         self.create_key(uid)