|
11 | 11 | ServiceListFilters, |
12 | 12 | ServiceUpdateV2, |
13 | 13 | ) |
| 14 | +from models_library.api_schemas_catalog.services_ports import ServicePortGet |
14 | 15 | from models_library.products import ProductName |
15 | 16 | from models_library.rest_pagination import PageOffsetInt |
16 | 17 | from models_library.rpc_pagination import DEFAULT_NUMBER_OF_ITEMS_PER_PAGE, PageLimitInt |
@@ -274,3 +275,43 @@ async def list_my_service_history_paginated( |
274 | 275 | offset=offset, |
275 | 276 | ), |
276 | 277 | ) |
| 278 | + |
| 279 | + |
| 280 | +@router.expose( |
| 281 | + reraise_if_error_type=( |
| 282 | + CatalogItemNotFoundError, |
| 283 | + CatalogForbiddenError, |
| 284 | + ValidationError, |
| 285 | + ) |
| 286 | +) |
| 287 | +@log_decorator(_logger, level=logging.DEBUG) |
| 288 | +@validate_call(config={"arbitrary_types_allowed": True}) |
| 289 | +async def get_service_ports( |
| 290 | + app: FastAPI, |
| 291 | + *, |
| 292 | + product_name: ProductName, |
| 293 | + user_id: UserID, |
| 294 | + service_key: ServiceKey, |
| 295 | + service_version: ServiceVersion, |
| 296 | +) -> list[ServicePortGet]: |
| 297 | + """Get service ports (inputs and outputs) for a specific service version""" |
| 298 | + assert app.state.engine # nosec |
| 299 | + |
| 300 | + service_ports = await catalog_services.get_user_services_ports( |
| 301 | + repo=ServicesRepository(app.state.engine), |
| 302 | + director_api=get_director_client(app), |
| 303 | + product_name=product_name, |
| 304 | + user_id=user_id, |
| 305 | + service_key=service_key, |
| 306 | + service_version=service_version, |
| 307 | + ) |
| 308 | + |
| 309 | + # Convert internal ServicePort models to ServicePortGet models for the API |
| 310 | + return [ |
| 311 | + ServicePortGet.from_domain_model( |
| 312 | + kind=port.kind, |
| 313 | + key=port.key, |
| 314 | + port=port.port, |
| 315 | + ) |
| 316 | + for port in service_ports |
| 317 | + ] |
0 commit comments