|
| 1 | +""" |
| 2 | +EntityLists resource for the MD Python v2 client. |
| 3 | +
|
| 4 | +Maps the two endpoints exposed under a workspace: |
| 5 | +
|
| 6 | + POST /api/workspaces/:workspace_id/entity_lists → create |
| 7 | + GET /api/workspaces/:workspace_id/entity_lists/:id → show |
| 8 | +
|
| 9 | +There is no list/index endpoint yet (the server-side controller exposes |
| 10 | +only Create and Show), so this resource intentionally does not implement |
| 11 | +``list``. Looking up an entity list requires its id. |
| 12 | +""" |
| 13 | + |
| 14 | +from typing import TYPE_CHECKING, Any, Dict, List, Optional, Sequence, Union |
| 15 | + |
| 16 | +from ...models.entity_list import EntityList, EntityListItem |
| 17 | + |
| 18 | +if TYPE_CHECKING: |
| 19 | + from ...base_client import BaseMDClient |
| 20 | + |
| 21 | + |
| 22 | +_JSON_HEADERS = {"Content-Type": "application/json"} |
| 23 | + |
| 24 | + |
| 25 | +def _check(response: Any, expected: int, action: str) -> None: |
| 26 | + if response.status_code != expected: |
| 27 | + raise Exception(f"Failed to {action}: {response.status_code} - {response.text}") |
| 28 | + |
| 29 | + |
| 30 | +# A caller can pass items as a list of EntityListItem objects or as a |
| 31 | +# list of plain dicts ({entity_id, group_id, dataset_id}). Both shapes |
| 32 | +# are normalised before sending. |
| 33 | +EntityListItemInput = Union[EntityListItem, Dict[str, Any]] |
| 34 | + |
| 35 | + |
| 36 | +class EntityLists: |
| 37 | + """Workspace-scoped entity lists (proteins / peptides / genes). |
| 38 | +
|
| 39 | + Reached via ``client.workspaces.entity_lists`` and always |
| 40 | + parameterised by ``workspace_id`` since lists live under a workspace. |
| 41 | + """ |
| 42 | + |
| 43 | + def __init__(self, client: "BaseMDClient"): |
| 44 | + self._client = client |
| 45 | + |
| 46 | + def _base(self, workspace_id: str) -> str: |
| 47 | + return f"/workspaces/{workspace_id}/entity_lists" |
| 48 | + |
| 49 | + def create( |
| 50 | + self, |
| 51 | + workspace_id: str, |
| 52 | + name: str, |
| 53 | + entity_type: str, |
| 54 | + items: Sequence[EntityListItemInput], |
| 55 | + ) -> EntityList: |
| 56 | + """Create a named entity list inside a workspace. |
| 57 | +
|
| 58 | + Args: |
| 59 | + workspace_id: Parent workspace UUID. |
| 60 | + name: Display name for the list. |
| 61 | + entity_type: One of ``protein``, ``peptide``, or ``gene``. |
| 62 | + items: At least one item. Each item is either an |
| 63 | + :class:`EntityListItem` or a dict with ``entity_id``, |
| 64 | + ``group_id`` and ``dataset_id``. |
| 65 | +
|
| 66 | + Returns: |
| 67 | + The created :class:`EntityList` (with ``items`` populated). |
| 68 | + """ |
| 69 | + if entity_type not in {"protein", "peptide", "gene"}: |
| 70 | + raise ValueError( |
| 71 | + "entity_type must be one of: protein, peptide, gene " |
| 72 | + f"(got {entity_type!r})" |
| 73 | + ) |
| 74 | + if not items: |
| 75 | + raise ValueError("items must include at least one entry") |
| 76 | + |
| 77 | + payload_items: List[Dict[str, Any]] = [] |
| 78 | + for item in items: |
| 79 | + if isinstance(item, EntityListItem): |
| 80 | + payload_items.append(item.to_create_payload()) |
| 81 | + elif isinstance(item, dict): |
| 82 | + if "entity_id" not in item: |
| 83 | + raise ValueError("every item must have an 'entity_id' field") |
| 84 | + payload_items.append(dict(item)) |
| 85 | + else: |
| 86 | + raise TypeError( |
| 87 | + "items entries must be EntityListItem or dict, " |
| 88 | + f"got {type(item).__name__}" |
| 89 | + ) |
| 90 | + |
| 91 | + response = self._client._make_request( |
| 92 | + method="POST", |
| 93 | + endpoint=self._base(workspace_id), |
| 94 | + json={ |
| 95 | + "name": name, |
| 96 | + "entity_type": entity_type, |
| 97 | + "items": payload_items, |
| 98 | + }, |
| 99 | + headers=_JSON_HEADERS, |
| 100 | + ) |
| 101 | + _check(response, 201, "create entity list") |
| 102 | + return EntityList.from_json(response.json()) |
| 103 | + |
| 104 | + def get(self, workspace_id: str, list_id: str) -> Optional[EntityList]: |
| 105 | + """Get a single entity list, or ``None`` if not found.""" |
| 106 | + response = self._client._make_request( |
| 107 | + method="GET", |
| 108 | + endpoint=f"{self._base(workspace_id)}/{list_id}", |
| 109 | + ) |
| 110 | + if response.status_code == 404: |
| 111 | + return None |
| 112 | + _check(response, 200, "get entity list") |
| 113 | + return EntityList.from_json(response.json()) |
0 commit comments