diff --git a/.gitlab-ci.yml b/.gitlab-ci.yml
index 447a9728f58d483f834e24ecc08b86e11f6a3040..aae2876ad6430eaccb3bb138e7aff588679faf97 100644
--- a/.gitlab-ci.yml
+++ b/.gitlab-ci.yml
@@ -1,4 +1,4 @@
-image: ${CI_DEPENDENCY_PROXY_DIRECT_GROUP_IMAGE_PREFIX}/python:3.11-slim
+image: ${CI_DEPENDENCY_PROXY_DIRECT_GROUP_IMAGE_PREFIX}/python:3.12-slim
variables:
PIP_CACHE_DIR: "$CI_PROJECT_DIR/.cache/pip"
diff --git a/.pre-commit-config.yaml b/.pre-commit-config.yaml
index 9b2bf6ac8d8d6a02e9785ce6c35ef7f348c59298..2e8793959ae1ebe78b898fe22fef6651b3bafabe 100644
--- a/.pre-commit-config.yaml
+++ b/.pre-commit-config.yaml
@@ -15,13 +15,13 @@ repos:
- id: check-merge-conflict
- id: check-ast
- repo: https://github.com/psf/black
- rev: 23.12.0
+ rev: 23.12.1
hooks:
- id: black
files: app
args: [--check]
- repo: https://github.com/charliermarsh/ruff-pre-commit
- rev: 'v0.1.8'
+ rev: 'v0.1.11'
hooks:
- id: ruff
- repo: https://github.com/PyCQA/isort
@@ -31,7 +31,7 @@ repos:
files: app
args: [-c]
- repo: https://github.com/pre-commit/mirrors-mypy
- rev: v1.7.1
+ rev: v1.8.0
hooks:
- id: mypy
files: app
diff --git a/Dockerfile b/Dockerfile
index d364a9b2efbb0abeb5800317b3544c8c059f56f7..ff7b1fbbf58222bf05d6aca64954f3e1244177b2 100644
--- a/Dockerfile
+++ b/Dockerfile
@@ -1,4 +1,4 @@
-FROM python:3.11-slim
+FROM python:3.12-slim
EXPOSE 8000
# dumb-init forwards the kill signal to the python process
diff --git a/app/api/endpoints/resource_version.py b/app/api/endpoints/resource_version.py
index ce923649aaa8ece90cb5a38bd3c22650dcf764e6..2da37ad792458bcd60e403aaf78389f70696325b 100644
--- a/app/api/endpoints/resource_version.py
+++ b/app/api/endpoints/resource_version.py
@@ -1,8 +1,9 @@
-from typing import Annotated, Any, Awaitable, Callable, List, Optional
+from typing import Annotated, Any, Awaitable, Callable, List, Union
from clowmdb.models import ResourceVersion
from fastapi import APIRouter, BackgroundTasks, Depends, HTTPException, Query, status
from opentelemetry import trace
+from pydantic.json_schema import SkipJsonSchema
from app.api.dependencies import (
AuthorizationDependency,
@@ -43,7 +44,7 @@ async def list_resource_versions(
resource: CurrentResource,
current_user: CurrentUser,
version_status: Annotated[
- Optional[List[ResourceVersion.Status]],
+ Union[List[ResourceVersion.Status], SkipJsonSchema[None]],
Query(
description=f"Which versions to include in the response. Permission `resource:read_any` required, current user is the maintainer, then only permission `resource:read` required. Default `{ResourceVersion.Status.LATEST.name}` and `{ResourceVersion.Status.SYNCHRONIZED.name}`.",
# noqa: E501
@@ -92,7 +93,6 @@ async def request_resource_version(
resource_version_in: ResourceVersionIn,
current_user: CurrentUser,
db: DBSession,
- s3: S3Resource,
background_tasks: BackgroundTasks,
) -> ResourceVersionOut:
"""
@@ -131,13 +131,11 @@ async def request_resource_version(
resource_version_out = ResourceVersionOut.from_db_resource_version(resource_version)
background_tasks.add_task(
give_permission_to_s3_resource_version,
- s3=s3,
resource_version=resource_version_out,
maintainer_id=current_user.uid,
)
background_tasks.add_task(
add_s3_resource_version_info,
- s3=s3,
s3_resource_version_info=S3ResourceVersionInfo.from_models(
resource=resource, resource_version=resource.versions[0], maintainer=current_user
),
@@ -233,7 +231,7 @@ async def request_resource_version_sync(
detail=f"Can't request sync for resource version with status {resource_version.status.name}",
)
resource_version_out = ResourceVersionOut.from_db_resource_version(resource_version)
- if await get_s3_resource_version_obj(s3, ResourceVersionOut.from_db_resource_version(resource_version)) is None:
+ if await get_s3_resource_version_obj(ResourceVersionOut.from_db_resource_version(resource_version), s3=s3) is None:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail=f"Missing resource at S3 path {resource_version_out.s3_path}",
@@ -244,7 +242,6 @@ async def request_resource_version_sync(
resource_version_out.status = ResourceVersion.Status.SYNC_REQUESTED
background_tasks.add_task(
remove_permission_to_s3_resource_version,
- s3=s3,
resource_version=resource_version_out,
)
return resource_version_out
@@ -296,7 +293,7 @@ async def resource_version_sync(
detail=f"Can't sync resource version with status {resource_version.status.name}",
)
resource_version_out = ResourceVersionOut.from_db_resource_version(resource_version)
- if await get_s3_resource_version_obj(s3, ResourceVersionOut.from_db_resource_version(resource_version)) is None:
+ if await get_s3_resource_version_obj(ResourceVersionOut.from_db_resource_version(resource_version), s3=s3) is None:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail=f"Missing resource at S3 path {resource_version_out.s3_path}",
@@ -306,9 +303,7 @@ async def resource_version_sync(
)
resource_version_out.status = ResourceVersion.Status.SYNCHRONIZING
- background_tasks.add_task(
- synchronize_cluster_resource, db=db, slurm_client=slurm_client, resource_version=resource_version_out
- )
+ background_tasks.add_task(synchronize_cluster_resource, resource_version=resource_version_out)
return resource_version_out
@@ -352,9 +347,7 @@ async def resource_version_latest(
detail=f"Can't set resource version to {ResourceVersion.Status.LATEST.name} with status {resource_version.status.name}",
)
resource_version_out = ResourceVersionOut.from_db_resource_version(resource_version)
- background_tasks.add_task(
- set_cluster_resource_version_latest, db=db, slurm_client=slurm_client, resource_version=resource_version_out
- )
+ background_tasks.add_task(set_cluster_resource_version_latest, resource_version=resource_version_out)
return resource_version_out
@@ -397,9 +390,7 @@ async def delete_resource_version_cluster(
)
resource_version.status = ResourceVersion.Status.CLUSTER_DELETED
resource_version_out = ResourceVersionOut.from_db_resource_version(resource_version)
- background_tasks.add_task(
- delete_cluster_resource_version, slurm_client=slurm_client, resource_version=resource_version_out
- )
+ background_tasks.add_task(delete_cluster_resource_version, resource_version=resource_version_out)
return resource_version_out
@@ -443,13 +434,11 @@ async def delete_resource_version_s3(
resource_version.status = ResourceVersion.Status.S3_DELETED
background_tasks.add_task(
delete_s3_resource_version,
- s3=s3,
resource_id=resource.resource_id,
resource_version_id=resource_version.resource_version_id,
)
background_tasks.add_task(
remove_permission_to_s3_resource_version,
- s3=s3,
resource_version=ResourceVersionOut.from_db_resource_version(resource_version),
)
diff --git a/app/api/endpoints/resources.py b/app/api/endpoints/resources.py
index 615566e639f644401b0307b12464a6ffb777b05a..a7d0987360ede0d69779e80fba14b273975cc43d 100644
--- a/app/api/endpoints/resources.py
+++ b/app/api/endpoints/resources.py
@@ -1,26 +1,19 @@
-from typing import Annotated, Any, Awaitable, Callable, List, Optional
+from typing import Annotated, Any, Awaitable, Callable, List, Union
from clowmdb.models import ResourceVersion
from fastapi import APIRouter, BackgroundTasks, Depends, Query, status
from opentelemetry import trace
+from pydantic.json_schema import SkipJsonSchema
-from app.api.dependencies import (
- AuthorizationDependency,
- CurrentResource,
- CurrentUser,
- DBSession,
- S3Resource,
- SlurmClient,
-)
+from app.api.dependencies import AuthorizationDependency, CurrentResource, CurrentUser, DBSession
from app.api.resource_cluster_utils import delete_cluster_resource
from app.api.resource_s3_utils import (
add_s3_resource_version_info,
- delete_s3_bucket_policy_stmt,
+ delete_resource_policy_stmt,
delete_s3_resource,
give_permission_to_s3_resource,
give_permission_to_s3_resource_version,
)
-from app.core.config import settings
from app.crud import CRUDResource
from app.schemas.resource import ResourceIn, ResourceOut, S3ResourceVersionInfo
from app.utils.otlp import start_as_current_span_async
@@ -38,7 +31,7 @@ async def list_resources(
current_user: CurrentUser,
db: DBSession,
maintainer_id: Annotated[
- Optional[str],
+ Union[str, SkipJsonSchema[None]],
Query(
max_length=64,
description="Filter for resource by maintainer. If current user is the same as maintainer ID, permission `resource:list` required, otherwise `resource:list_filter`.",
@@ -46,13 +39,13 @@ async def list_resources(
),
] = None,
version_status: Annotated[
- Optional[List[ResourceVersion.Status]],
+ Union[List[ResourceVersion.Status], SkipJsonSchema[None]],
Query(
description=f"Which versions of the resource to include in the response. Permission `resource:list_filter` required, unless `maintainer_id` is provided and current user is maintainer, then only permission `resource:list` required. Default `{ResourceVersion.Status.LATEST.name}` and `{ResourceVersion.Status.SYNCHRONIZED.name}`.",
# noqa: E501
),
] = None,
- name_substring: Annotated[Optional[str], Query(max_length=32)] = None,
+ name_substring: Annotated[Union[str, SkipJsonSchema[None]], Query(max_length=32)] = None,
) -> List[ResourceOut]:
"""
List all resources.
@@ -106,7 +99,6 @@ async def request_resource(
current_user: CurrentUser,
resource_in: ResourceIn,
db: DBSession,
- s3: S3Resource,
background_tasks: BackgroundTasks,
) -> ResourceOut:
"""
@@ -135,16 +127,14 @@ async def request_resource(
await authorization("create")
resource = await CRUDResource.create(db, resource_in=resource_in, maintainer_id=current_user.uid)
resource_out = ResourceOut.from_db_resource(db_resource=resource)
- background_tasks.add_task(give_permission_to_s3_resource, s3=s3, resource=resource_out)
+ background_tasks.add_task(give_permission_to_s3_resource, resource=resource_out)
background_tasks.add_task(
give_permission_to_s3_resource_version,
- s3=s3,
resource_version=resource_out.versions[0],
maintainer_id=resource_out.maintainer_id,
)
background_tasks.add_task(
add_s3_resource_version_info,
- s3=s3,
s3_resource_version_info=S3ResourceVersionInfo.from_models(
resource=resource, resource_version=resource.versions[0], maintainer=current_user
),
@@ -159,7 +149,7 @@ async def get_resource(
resource: CurrentResource,
current_user: CurrentUser,
version_status: Annotated[
- Optional[List[ResourceVersion.Status]],
+ Union[List[ResourceVersion.Status], SkipJsonSchema[None]],
Query(
description=f"Which versions of the resource to include in the response. Permission `resource:read_any` required, unless the current user is the maintainer, then only permission `resource:read` required. Default `{ResourceVersion.Status.LATEST.name}` and `{ResourceVersion.Status.SYNCHRONIZED.name}`.",
# noqa: E501
@@ -202,9 +192,7 @@ async def delete_resource(
authorization: Authorization,
resource: CurrentResource,
db: DBSession,
- s3: S3Resource,
background_tasks: BackgroundTasks,
- slurm_client: SlurmClient,
) -> None:
"""
Delete a resources.
@@ -223,18 +211,10 @@ async def delete_resource(
Entrypoint for new BackgroundTasks. Provided by FastAPI.
s3 : types_aiobotocore_s3.service_resource import S3ServiceResource
S3 Service to perform operations on buckets. Dependency Injection.
- slurm_client : app.slurm.rest_client.SlurmClient
- Slurm client with an open connection. Dependency Injection
"""
trace.get_current_span().set_attribute("resource_id", str(resource.resource_id))
await authorization("delete")
await CRUDResource.delete(db, resource_id=resource.resource_id)
- background_tasks.add_task(
- delete_s3_bucket_policy_stmt,
- s3=s3,
- bucket_name=settings.RESOURCE_BUCKET,
- sid=[str(resource.resource_id)]
- + [str(resource_version.resource_version_id) for resource_version in resource.versions],
- )
- background_tasks.add_task(delete_s3_resource, s3=s3, resource_id=resource.resource_id)
- background_tasks.add_task(delete_cluster_resource, slurm_client=slurm_client, resource_id=resource.resource_id)
+ background_tasks.add_task(delete_resource_policy_stmt, resource=ResourceOut.from_db_resource(resource))
+ background_tasks.add_task(delete_s3_resource, resource_id=resource.resource_id)
+ background_tasks.add_task(delete_cluster_resource, resource_id=resource.resource_id)
diff --git a/app/api/resource_cluster_utils.py b/app/api/resource_cluster_utils.py
index 6bc92257370584d643bd42539fce0f9f33955939..e3e11b1e4fc191ae0953df9e31fedc44d69cd00b 100644
--- a/app/api/resource_cluster_utils.py
+++ b/app/api/resource_cluster_utils.py
@@ -1,14 +1,14 @@
from asyncio import sleep as async_sleep
from pathlib import Path
-from typing import Optional
+from typing import Any, AsyncGenerator, Optional
from uuid import UUID
from clowmdb.models import ResourceVersion
-from httpx import HTTPError
+from httpx import AsyncClient, HTTPError
from mako.template import Template
from opentelemetry import trace
-from sqlalchemy.ext.asyncio import AsyncSession
+from app.api.dependencies import get_db, get_slurm_client
from app.core.config import settings
from app.crud import CRUDResourceVersion
from app.schemas.resource_version import ResourceVersionOut, resource_dir_name, resource_version_dir_name
@@ -23,9 +23,17 @@ delete_resource_version_script_template = Template(filename="app/mako_templates/
tracer = trace.get_tracer_provider().get_tracer(__name__)
-async def synchronize_cluster_resource(
- db: AsyncSession, slurm_client: SlurmClient, resource_version: ResourceVersionOut
-) -> None:
+async def get_async_http_client() -> AsyncGenerator[AsyncClient, None]:
+ async with AsyncClient() as client:
+ yield client
+
+
+async def update_db_resource_wrapper(**kwargs: Any) -> None:
+ async for db in get_db():
+ await CRUDResourceVersion.update_status(db=db, **kwargs)
+
+
+async def synchronize_cluster_resource(resource_version: ResourceVersionOut) -> None:
"""
Synchronize a resource to the cluster
@@ -33,19 +41,15 @@ async def synchronize_cluster_resource(
----------
db : sqlalchemy.ext.asyncio.AsyncSession.
Async database session to perform query on.
- slurm_client : app.slurm.rest_client.SlurmClient
- Slurm Rest Client to communicate with Slurm cluster.
resource_version : app.schemas.resource_version.ResourceVersionOut
Resource version schema to synchronize.
"""
-
synchronization_script = synchronization_script_template.render(
s3_url=settings.OBJECT_GATEWAY_URI,
resource_version_s3_folder=f"{settings.RESOURCE_BUCKET}/{resource_version_dir_name(resource_version.resource_id, resource_version.resource_version_id)}",
)
failed_job = AsyncJob(
- CRUDResourceVersion.update_status,
- db=db,
+ update_db_resource_wrapper,
resource_version_id=resource_version.resource_version_id,
status=ResourceVersion.Status.SYNC_REQUESTED,
resource_id=None,
@@ -71,26 +75,26 @@ async def synchronize_cluster_resource(
},
},
)
- # Try to start the job on the slurm cluster
- slurm_job_id = await slurm_client.submit_job(job_submission=job_submission)
- await _monitor_proper_job_execution(
- slurm_client=slurm_client,
- slurm_job_id=slurm_job_id,
- success_job=AsyncJob(
- CRUDResourceVersion.update_status,
- db=db,
- resource_version_id=resource_version.resource_version_id,
- status=ResourceVersion.Status.SYNCHRONIZED,
- resource_id=None,
- ),
- failed_job=failed_job,
- )
+ async for client in get_async_http_client():
+ slurm_client = get_slurm_client(client)
+ # Try to start the job on the slurm cluster
+ slurm_job_id = await slurm_client.submit_job(job_submission=job_submission)
+ await _monitor_proper_job_execution(
+ slurm_client=slurm_client,
+ slurm_job_id=slurm_job_id,
+ success_job=AsyncJob(
+ update_db_resource_wrapper,
+ resource_version_id=resource_version.resource_version_id,
+ status=ResourceVersion.Status.SYNCHRONIZED,
+ resource_id=None,
+ ),
+ failed_job=failed_job,
+ )
except (HTTPError, KeyError): # pragma: no cover
await failed_job()
async def delete_cluster_resource(
- slurm_client: SlurmClient,
resource_id: UUID,
) -> None:
"""
@@ -98,8 +102,6 @@ async def delete_cluster_resource(
Parameters
----------
- slurm_client : app.slurm.rest_client.SlurmClient
- Slurm Rest Client to communicate with Slurm cluster.
resource_id : uuid.UUID
ID of the resource to delete
"""
@@ -119,25 +121,21 @@ async def delete_cluster_resource(
},
},
)
- # Try to start the job on the slurm cluster
- slurm_job_id = await slurm_client.submit_job(job_submission=job_submission)
- await _monitor_proper_job_execution(slurm_client=slurm_client, slurm_job_id=slurm_job_id)
+ async for client in get_async_http_client():
+ slurm_client = get_slurm_client(client)
+ # Try to start the job on the slurm cluster
+ slurm_job_id = await slurm_client.submit_job(job_submission=job_submission)
+ await _monitor_proper_job_execution(slurm_client=slurm_client, slurm_job_id=slurm_job_id)
except (HTTPError, KeyError): # pragma: no cover
pass
-async def set_cluster_resource_version_latest(
- db: AsyncSession, slurm_client: SlurmClient, resource_version: ResourceVersionOut
-) -> None:
+async def set_cluster_resource_version_latest(resource_version: ResourceVersionOut) -> None:
"""
Synchronize a resource to the cluster
Parameters
----------
- db : sqlalchemy.ext.asyncio.AsyncSession.
- Async database session to perform query on.
- slurm_client : app.slurm.rest_client.SlurmClient
- Slurm Rest Client to communicate with Slurm cluster.
resource_version : app.schemas.resource_version.ResourceVersionOut
Resource version schema to synchronize.
"""
@@ -173,26 +171,26 @@ async def set_cluster_resource_version_latest(
},
},
)
- # Try to start the job on the slurm cluster
- slurm_job_id = await slurm_client.submit_job(job_submission=job_submission)
+ async for client in get_async_http_client():
+ slurm_client = get_slurm_client(client)
+ # Try to start the job on the slurm cluster
+ slurm_job_id = await slurm_client.submit_job(job_submission=job_submission)
- await _monitor_proper_job_execution(
- slurm_client=slurm_client,
- slurm_job_id=slurm_job_id,
- success_job=AsyncJob(
- CRUDResourceVersion.update_status,
- db=db,
- status=ResourceVersion.Status.LATEST,
- resource_version_id=resource_version.resource_version_id,
- resource_id=resource_version.resource_id,
- ),
- )
+ await _monitor_proper_job_execution(
+ slurm_client=slurm_client,
+ slurm_job_id=slurm_job_id,
+ success_job=AsyncJob(
+ update_db_resource_wrapper,
+ status=ResourceVersion.Status.LATEST,
+ resource_version_id=resource_version.resource_version_id,
+ resource_id=resource_version.resource_id,
+ ),
+ )
except (HTTPError, KeyError): # pragma: no cover
pass
async def delete_cluster_resource_version(
- slurm_client: SlurmClient,
resource_version: ResourceVersionOut,
) -> None:
"""
@@ -200,8 +198,6 @@ async def delete_cluster_resource_version(
Parameters
----------
- slurm_client : app.slurm.rest_client.SlurmClient
- Slurm Rest Client to communicate with Slurm cluster.
resource_version : app.schemas.resource_version.ResourceVersionOut
Resource version schema to delete.
"""
@@ -237,9 +233,11 @@ async def delete_cluster_resource_version(
},
},
)
- # Try to start the job on the slurm cluster
- slurm_job_id = await slurm_client.submit_job(job_submission=job_submission)
- await _monitor_proper_job_execution(slurm_client=slurm_client, slurm_job_id=slurm_job_id)
+ async for client in get_async_http_client():
+ slurm_client = get_slurm_client(client)
+ # Try to start the job on the slurm cluster
+ slurm_job_id = await slurm_client.submit_job(job_submission=job_submission)
+ await _monitor_proper_job_execution(slurm_client=slurm_client, slurm_job_id=slurm_job_id)
except (HTTPError, KeyError): # pragma: no cover
pass
diff --git a/app/api/resource_s3_utils.py b/app/api/resource_s3_utils.py
index 4e274d2adcee8d3fd3f5e409a69855aab11a7240..c6a0c60509a4d2fecc8ad486653200fe0f9336cd 100644
--- a/app/api/resource_s3_utils.py
+++ b/app/api/resource_s3_utils.py
@@ -1,5 +1,5 @@
from io import BytesIO
-from typing import TYPE_CHECKING, Any, Dict, Optional
+from typing import TYPE_CHECKING, Any, AsyncGenerator, Dict, Optional
from uuid import UUID
from opentelemetry import trace
@@ -7,6 +7,7 @@ from opentelemetry import trace
from app.core.config import settings
from app.s3.s3_resource import (
add_s3_bucket_policy_stmt,
+ boto_session,
delete_s3_bucket_policy_stmt,
delete_s3_objects,
get_s3_object,
@@ -30,14 +31,21 @@ else:
tracer = trace.get_tracer_provider().get_tracer(__name__)
-async def give_permission_to_s3_resource(s3: S3ServiceResource, resource: ResourceOut) -> None:
+async def get_s3_resource() -> AsyncGenerator[S3ServiceResource, None]:
+ async with boto_session.resource(
+ service_name="s3",
+ endpoint_url=str(settings.OBJECT_GATEWAY_URI)[:-1],
+ verify=str(settings.OBJECT_GATEWAY_URI).startswith("https"),
+ ) as s3_resource:
+ yield s3_resource
+
+
+async def give_permission_to_s3_resource(resource: ResourceOut) -> None:
"""
Give the maintainer permissions to list the S3 objects of his resource.
Parameters
----------
- s3 : types_aiobotocore_s3.service_resource import S3ServiceResource
- S3 Service to perform operations on buckets.
resource : app.schemas.resource.ResourceOut
Information about the resource.
"""
@@ -49,19 +57,16 @@ async def give_permission_to_s3_resource(s3: S3ServiceResource, resource: Resour
"Action": ["s3:ListBucket"],
"Condition": {"StringLike": {"s3:prefix": resource_dir_name(resource.resource_id) + "/*"}},
}
- await add_s3_bucket_policy_stmt(s3=s3, bucket_name=settings.RESOURCE_BUCKET, stmt=policy)
+ async for s3 in get_s3_resource():
+ await add_s3_bucket_policy_stmt(s3=s3, bucket_name=settings.RESOURCE_BUCKET, stmt=policy)
-async def give_permission_to_s3_resource_version(
- s3: S3ServiceResource, resource_version: ResourceVersionOut, maintainer_id: str
-) -> None:
+async def give_permission_to_s3_resource_version(resource_version: ResourceVersionOut, maintainer_id: str) -> None:
"""
Give the maintainer permissions to upload the resource to the appropriate S3 key.
Parameters
----------
- s3 : types_aiobotocore_s3.service_resource import S3ServiceResource
- S3 Service to perform operations on buckets.
resource_version : app.schemas.resource_version.ResourceVersionOut
Information about the resource version.
maintainer_id : str
@@ -74,10 +79,11 @@ async def give_permission_to_s3_resource_version(
"Resource": f"arn:aws:s3:::{resource_version.s3_path[5:]}",
"Action": ["s3:DeleteObject", "s3:PutObject"],
}
- await add_s3_bucket_policy_stmt(s3=s3, bucket_name=settings.RESOURCE_BUCKET, stmt=policy)
+ async for s3 in get_s3_resource():
+ await add_s3_bucket_policy_stmt(s3=s3, bucket_name=settings.RESOURCE_BUCKET, stmt=policy)
-async def add_s3_resource_version_info(s3: S3ServiceResource, s3_resource_version_info: S3ResourceVersionInfo) -> None:
+async def add_s3_resource_version_info(s3_resource_version_info: S3ResourceVersionInfo) -> None:
"""
Upload the resource version information to S3 for documentation
@@ -89,85 +95,109 @@ async def add_s3_resource_version_info(s3: S3ServiceResource, s3_resource_versio
Resource information object that will be uploaded to S3.
"""
buf = BytesIO(s3_resource_version_info.model_dump_json(indent=2).encode("utf-8"))
- await upload_obj(
- s3=s3,
- bucket_name=settings.RESOURCE_BUCKET,
- key=s3_resource_version_info.s3_path(),
- obj_stream=buf,
- ExtraArgs={"ContentType": "application/json"},
- )
+ async for s3 in get_s3_resource():
+ await upload_obj(
+ s3=s3,
+ bucket_name=settings.RESOURCE_BUCKET,
+ key=s3_resource_version_info.s3_path(),
+ obj_stream=buf,
+ ExtraArgs={"ContentType": "application/json"},
+ )
-async def remove_permission_to_s3_resource_version(s3: S3ServiceResource, resource_version: ResourceVersionOut) -> None:
+async def remove_permission_to_s3_resource_version(resource_version: ResourceVersionOut) -> None:
"""
Remove the permission of the maintainer to upload a resource to S3.
Parameters
----------
- s3 : types_aiobotocore_s3.service_resource import S3ServiceResource
- S3 Service to perform operations on buckets.
resource_version : app.schemas.resource_version.ResourceVersionOut
Information about the resource version.
"""
- await delete_s3_bucket_policy_stmt(
- s3, bucket_name=settings.RESOURCE_BUCKET, sid=str(resource_version.resource_version_id)
- )
+ async for s3 in get_s3_resource():
+ await delete_s3_bucket_policy_stmt(
+ s3, bucket_name=settings.RESOURCE_BUCKET, sid=str(resource_version.resource_version_id)
+ )
-async def delete_s3_resource(s3: S3ServiceResource, resource_id: UUID) -> None:
+async def delete_s3_resource(resource_id: UUID) -> None:
"""
Delete all objects related to a resource in S3.
Parameters
----------
- s3 : types_aiobotocore_s3.service_resource import S3ServiceResource
- S3 Service to perform operations on buckets.
resource_id : uuid.UUID
ID of the resource
"""
- await delete_s3_objects(s3, bucket_name=settings.RESOURCE_BUCKET, prefix=resource_dir_name(resource_id) + "/")
+ async for s3 in get_s3_resource():
+ await delete_s3_objects(s3, bucket_name=settings.RESOURCE_BUCKET, prefix=resource_dir_name(resource_id) + "/")
-async def delete_s3_resource_version(s3: S3ServiceResource, resource_id: UUID, resource_version_id: UUID) -> None:
+async def delete_s3_resource_version(resource_id: UUID, resource_version_id: UUID) -> None:
"""
Delete all objects related to a resource version in S3.
Parameters
----------
- s3 : types_aiobotocore_s3.service_resource import S3ServiceResource
- S3 Service to perform operations on buckets.
resource_id : uuid.UUID
ID of the resource
resource_version_id : uuid.UUID
ID of the resource version
"""
- await delete_s3_objects(
- s3, bucket_name=settings.RESOURCE_BUCKET, prefix=resource_version_dir_name(resource_id, resource_version_id)
- )
+ async for s3 in get_s3_resource():
+ await delete_s3_objects(
+ s3, bucket_name=settings.RESOURCE_BUCKET, prefix=resource_version_dir_name(resource_id, resource_version_id)
+ )
async def get_s3_resource_version_obj(
- s3: S3ServiceResource, resource_version: ResourceVersionOut
+ resource_version: ResourceVersionOut,
+ s3: Optional[S3ServiceResource] = None,
) -> Optional[ObjectSummary]:
"""
Delete all objects related to a resource version in S3.
Parameters
----------
- s3 : types_aiobotocore_s3.service_resource import S3ServiceResource
- S3 Service to perform operations on buckets.
resource_version : app.schemas.resource_version.ResourceVersionOut
Information about the resource version.
+ s3 : types_aiobotocore_s3.service_resource import S3ServiceResource | None
+ S3 Service to perform operations on buckets. If None, a new connection will be established.
Returns
-------
obj : types_aiobotocore_s3.service_resource.ObjectSummary | None
The object summary of the resource version if it exists.
"""
- return await get_s3_object(
- s3=s3,
- bucket_name=settings.RESOURCE_BUCKET,
- key=resource_version_key(
- resource_id=resource_version.resource_id, resource_version_id=resource_version.resource_version_id
- ),
- )
+
+ async def func(s3_inner: S3ServiceResource) -> Optional[ObjectSummary]:
+ return await get_s3_object(
+ s3=s3_inner,
+ bucket_name=settings.RESOURCE_BUCKET,
+ key=resource_version_key(
+ resource_id=resource_version.resource_id, resource_version_id=resource_version.resource_version_id
+ ),
+ )
+
+ if s3 is None:
+ async for s3_outer in get_s3_resource():
+ return await func(s3_outer)
+ return await func(s3) # type: ignore[arg-type]
+
+
+async def delete_resource_policy_stmt(resource: ResourceOut) -> None:
+ """
+ Delete all bucket policy statements regarding ths resource.
+
+ Parameters
+ ----------
+ resource : app.schemas.resource.ResourceOut
+ Information about the resource.
+ """
+ async for s3 in get_s3_resource():
+ await delete_s3_bucket_policy_stmt(
+ s3=s3,
+ bucket_name=settings.RESOURCE_BUCKET,
+ sid=[str(resource.resource_id)]
+ + [str(resource_version.resource_version_id) for resource_version in resource.versions],
+ )
diff --git a/app/tests/conftest.py b/app/tests/conftest.py
index 57daa41f575025d99988a985440c507e305d85b4..9b9e28dfc51dc6af50a6ef1e9b7d874f9553c45f 100644
--- a/app/tests/conftest.py
+++ b/app/tests/conftest.py
@@ -14,6 +14,7 @@ from pytrie import SortedStringTrie as Trie
from sqlalchemy import delete, update
from sqlalchemy.ext.asyncio import AsyncSession
+from app.api import dependencies, resource_cluster_utils, resource_s3_utils
from app.api.dependencies import get_db, get_decode_jwt_function, get_httpx_client, get_s3_resource
from app.core.config import settings
from app.main import app
@@ -64,6 +65,52 @@ def mock_opa_service() -> Iterator[MockOpaService]:
mock_opa.reset()
+@pytest_asyncio.fixture(scope="session")
+async def mock_client(
+ mock_opa_service: MockOpaService,
+ mock_slurm_cluster: MockSlurmCluster,
+) -> AsyncIterator[httpx.AsyncClient]:
+ endpoints: Dict[str, MockHTTPService] = {
+ str(settings.SLURM_ENDPOINT): mock_slurm_cluster,
+ str(settings.OPA_URI): mock_opa_service,
+ }
+ # data structure to easily find the appropriate mock request based on the URL
+ t = Trie(**endpoints)
+
+ def mock_request_handler(request: httpx.Request) -> httpx.Response:
+ url = str(request.url)
+ return t.longest_prefix_value(url, DefaultMockHTTPService()).handle_request(request)
+
+ async with httpx.AsyncClient(transport=httpx.MockTransport(mock_request_handler)) as http_client:
+ yield http_client
+
+
+@pytest.fixture(autouse=True)
+def monkeypatch_background_tasks(
+ monkeypatch: pytest.MonkeyPatch,
+ mock_client: httpx.AsyncClient,
+ db: AsyncSession,
+ mock_s3_service: MockS3ServiceResource,
+) -> None:
+ """
+ Patch the functions to get resources in background tasks with mock resources.
+ """
+
+ async def get_http_client() -> AsyncIterator[httpx.AsyncClient]:
+ yield mock_client
+
+ async def get_patch_db() -> AsyncIterator[AsyncSession]:
+ yield db
+
+ async def get_s3() -> AsyncIterator[MockS3ServiceResource]:
+ yield mock_s3_service
+
+ monkeypatch.setattr(dependencies, "get_db", get_patch_db)
+ monkeypatch.setattr(resource_s3_utils, "get_s3_resource", get_s3)
+ monkeypatch.setattr(resource_cluster_utils, "get_async_http_client", get_http_client)
+ monkeypatch.setenv("OTLP_GRPC_ENDPOINT", "")
+
+
@pytest_asyncio.fixture(scope="module")
async def client(
mock_s3_service: MockS3ServiceResource,
diff --git a/pyproject.toml b/pyproject.toml
index e48e191ed42e52535b20dea692f0263c77c27b55..c1825ab66eb937ecd4857c45c036dadee5c9f29c 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -8,7 +8,7 @@ line-length = 120
[tool.ruff]
line-length = 120
-target-version = "py311"
+target-version = "py312"
[tool.mypy]
plugins = ["pydantic.mypy", "sqlalchemy.ext.mypy.plugin"]
diff --git a/requirements-dev.txt b/requirements-dev.txt
index 174bb00e549b7d9625c6a149ca30934f701a1477..83ce2a007d6bc39942bd10497f64e2173d4c6642 100644
--- a/requirements-dev.txt
+++ b/requirements-dev.txt
@@ -2,14 +2,14 @@
pytest>=7.4.0,<7.5.0
pytest-asyncio>=0.21.0,<0.22.0
pytest-cov>=4.1.0,<4.2.0
-coverage[toml]>=7.3.0,<7.4.0
+coverage[toml]>=7.4.0,<7.5.0
# Linters
ruff>=0.1.0,<0.2.0
black>=23.12.0,<24.0.0
isort>=5.13.0,<5.14.0
-mypy>=1.7.0,<1.8.0
+mypy>=1.8.0,<1.9.0
# stubs for mypy
-types-aiobotocore-lite[s3]>=2.8.0,<2.9.0
+types-aiobotocore-lite[s3]>=2.9.0,<2.10.0
types-requests
# Miscellaneous
pre-commit>=3.6.0,<3.7.0
diff --git a/requirements.txt b/requirements.txt
index 9b690e87b5fc1394d0da433925b5c64bca20c83a..550c2e6084e4aea31c2990fb80939b3ac58d841b 100644
--- a/requirements.txt
+++ b/requirements.txt
@@ -2,23 +2,22 @@
clowmdb>=2.3.0,<2.4.0
# Webserver packages
-anyio>=3.7.0,<4.0.0
-fastapi>=0.105.0,<0.106.0
+fastapi>=0.108.0,<0.109.0
pydantic>=2.5.0,<2.6.0
pydantic-settings>=2.1.0,<2.2.0
-uvicorn>=0.24.0,<0.25.0
+uvicorn>=0.25.0,<0.26.0
python-multipart
# Database packages
PyMySQL>=1.1.0,<1.2.0
SQLAlchemy>=2.0.0,<2.1.0
aiomysql>=0.2.0,<0.3.0
# Security packages
-authlib>=1.2.0,<1.3.0
+authlib>=1.3.0,<1.4.0
# S3 packages
aioboto3>=12.0.0,<13.0.0
# Miscellaneous
tenacity>=8.2.0,<8.3.0
-httpx>=0.25.0,<0.26.0
+httpx>=0.26.0,<0.27.0
itsdangerous
# template engine
mako>=1.3.0,<1.4.0