bec/bec_atlas
0
0
Fork 0
mirror of https://github.com/bec-project/bec_atlas.git synced 2025-07-13 22:51:49 +02:00
Code Issues Packages Projects Releases Wiki Activity

feat: added endpoint for retrieving realms and deployments but only for those with deployment access

Browse Source
This commit is contained in:
wakonig_k
2025-02-10 09:27:35 +01:00
parent 61b72953b6
commit 4d98c098cd
1 changed files with 43 additions and 1 deletions
Download Patch File Download Diff File Expand all files Collapse all files

44
backend/bec_atlas/router/realm_router.py
View File

@ -2,7 +2,7 @@ from fastapi import APIRouter, Depends
from bec_atlas.authentication import get_current_user
from bec_atlas.datasources.mongodb.mongodb import MongoDBDatasource
from bec_atlas.model.model import Realm, UserInfo
from bec_atlas.model.model import DeploymentAccess, Realm, UserInfo
from bec_atlas.router.base_router import BaseRouter
@ -27,6 +27,14 @@ class RealmRouter(BaseRouter):
response_model=Realm,
response_model_exclude_none=True,
)
self.router.add_api_route(
"/realms/deployment_access",
self.realm_with_deployment_access,
methods=["GET"],
description="Get all realms with deployment access",
response_model=list[Realm],
response_model_exclude_none=True,
)
async def realms(
self, include_deployments: bool = False, current_user: UserInfo = Depends(get_current_user)
@ -54,6 +62,40 @@ class RealmRouter(BaseRouter):
return self.db.aggregate("realms", include, Realm, user=current_user)
return self.db.find("realms", {}, Realm, user=current_user)
async def realm_with_deployment_access(
self, owner_only: bool = False, current_user: UserInfo = Depends(get_current_user)
):
"""
Get all realms with deployment access.
Args:
owner_only (bool): Only return realms where the current user is an owner of a deployment
current_user (UserInfo): The current user
Returns:
list[Realm]: List of realms with deployment access
"""
if owner_only:
access = self.db.find("deployment_access", {}, DeploymentAccess, user=current_user)
else:
access = self.db.find("deployment_access", {}, DeploymentAccess)
deployment_ids = [d.id for d in access]
include = [
{
"$lookup": {
"from": "deployments",
"let": {"realm_id": "$_id"},
"pipeline": [
{"$match": {"$expr": {"$eq": ["$realm_id", "$$realm_id"]}}},
{"$match": {"_id": {"$in": deployment_ids}}},
],
"as": "deployments",
}
},
{"$match": {"deployments": {"$ne": []}}},
]
return self.db.aggregate("realms", include, Realm, user=current_user)
async def realm_with_id(
self, realm_id: str, current_user: UserInfo = Depends(get_current_user)
):