enpi_api.l2.client.api.enrichment_api

View Source

  1import os
  2import zipfile
  3from pathlib import Path
  4
  5import pandas as pd
  6from loguru import logger
  7
  8from enpi_api.l1 import openapi_client
  9from enpi_api.l2.client.api.file_api import FileApi
 10from enpi_api.l2.events.workflow_execution_task_waitable import WorkflowExecutionTaskWaitable
 11from enpi_api.l2.types.api_error import ApiErrorContext
 12from enpi_api.l2.types.cluster import ClusterRunId
 13from enpi_api.l2.types.enrichment import (
 14    EnrichmentExportMode,
 15    EnrichmentRun,
 16    EnrichmentRunId,
 17    EnrichmentTemplate,
 18    EnrichmentTemplateId,
 19    EnrichmentTemplateOperation,
 20    EnrichmentWorkInput,
 21    SimplifiedEnrichmentTemplate,
 22    transform_operation,
 23    transform_operation_input,
 24)
 25from enpi_api.l2.types.execution import Execution
 26from enpi_api.l2.types.log import LogLevel
 27from enpi_api.l2.types.tag import TagId
 28from enpi_api.l2.types.task import TaskState
 29from enpi_api.l2.types.workflow import WorkflowExecutionId, WorkflowExecutionTaskId, WorkflowTaskTemplateName
 30from enpi_api.l2.util.file import unique_temp_dir
 31
 32
 33class EnrichmentApi:
 34    _inner_api_client: openapi_client.ApiClient
 35    _log_level: LogLevel
 36
 37    def __init__(self, inner_api_client: openapi_client.ApiClient, log_level: LogLevel):
 38        """@private"""
 39        self._inner_api_client = inner_api_client
 40        self._log_level = log_level
 41
 42    def get_runs(
 43        self,
 44    ) -> list[EnrichmentRun]:
 45        """Get all successful Enrichment Runs.
 46
 47        Returns:
 48            list[enpi_api.l2.types.enrichment.EnrichmentRun]: List of Enrichment Runs.
 49
 50        Raises:
 51            enpi_api.l2.types.api_error.ApiError: If API request fails.
 52
 53        Example:
 54            ```python
 55            with EnpiApiClient() as enpi_client:
 56                enrichment_runs = enpi_client.enrichment_api.get_runs()
 57            ```
 58        """
 59        enrichment_api_instance = openapi_client.EnrichmentApi(self._inner_api_client)
 60
 61        with ApiErrorContext():
 62            data = enrichment_api_instance.get_enrichment_runs()
 63
 64        return [EnrichmentRun.from_raw(cr) for cr in data.runs]
 65
 66    def get_run(self, enrichment_run_id: EnrichmentRunId) -> EnrichmentRun:
 67        """Get a single Enrichment Run by its ID.
 68
 69        Args:
 70            enrichment_run_id (enpi_api.l2.types.enrichment.EnrichmentRunId): ID of the Enrichment run to get.
 71
 72        Returns:
 73            enpi_api.l2.types.enrichment.EnrichmentRun: A successful Enrichment Run.
 74
 75        Raises:
 76            enpi_api.l2.types.api_error.ApiError: If API request fails.
 77
 78        Example:
 79            ```python
 80            with EnpiApiClient() as enpi_client:
 81                enrichment_run = enpi_client.enrichment_api.get_run(EnrichmentRunId(123))
 82            ```
 83        """
 84        enrichment_api_instance = openapi_client.EnrichmentApi(self._inner_api_client)
 85
 86        with ApiErrorContext():
 87            data = enrichment_api_instance.get_enrichment_run(int(enrichment_run_id))
 88
 89        return EnrichmentRun.from_raw(data.run)
 90
 91    def get_run_by_task_id(self, task_id: WorkflowExecutionTaskId) -> EnrichmentRun:
 92        """Get a single Enrichment Run by its task ID.
 93
 94        Args:
 95            task_id (enpi_api.l2.types.task.TaskId): ID of a task linked to a successful Enrichment Run.
 96
 97        Returns:
 98            enpi_api.l2.types.enrichment.EnrichmentRun: Successful Enrichment Run linked to the provided task ID.
 99
100        Raises:
101            enpi_api.l2.types.api_error.ApiError: If API request fails.
102
103        Example:
104            Fetch a Enrichment Run
105
106            ```python
107            with EnpiApiClient() as enpi_client:
108                enrichment_run = enpi_client.enrichment_api.get_run_by_task_id(TaskId(1234))
109            ```
110        """
111        enrichment_api_instance = openapi_client.EnrichmentApi(self._inner_api_client)
112
113        with ApiErrorContext():
114            data = enrichment_api_instance.get_enrichment_run_by_task_id(task_id)
115
116        return EnrichmentRun.from_raw(data.run)
117
118    def get_templates(self) -> list[SimplifiedEnrichmentTemplate]:
119        """Get all available Enrichment templates.
120
121        Returns:
122            list[enpi_api.l2.types.enrichment.SimplifiedEnrichmentTemplate]: Available Enrichment templates.
123
124        Raises:
125            enpi_api.l2.types.api_error.ApiError: If API request fails.
126
127        Example:
128            ```python
129            with EnpiApiClient() as enpi_client:
130                templates = enpi_client.enrichment_api.get_templates()
131            ```
132        """
133        enrichment_api_instance = openapi_client.EnrichmentApi(self._inner_api_client)
134
135        with ApiErrorContext():
136            data = enrichment_api_instance.get_enrichment_templates()
137
138        return [
139            SimplifiedEnrichmentTemplate(
140                id=EnrichmentTemplateId(d.id),
141                name=d.name,
142                created_at=d.created_at,
143            )
144            for d in data.templates
145        ]
146
147    def get_template(self, enrichment_template_id: EnrichmentTemplateId) -> EnrichmentTemplate:
148        """Get a Enrichment template by its ID.
149
150        Args:
151            enrichment_template_id (enpi_api.l2.types.enrichment.EnrichmentTemplateId): ID of a Enrichment template to get.
152
153        Returns:
154            enpi_api.l2.types.enrichment.EnrichmentTemplate: A Enrichment template matching the provided ID.
155
156        Raises:
157            enpi_api.l2.types.api_error.ApiError: If API request fails.
158
159        Example:
160            ```python
161            with EnpiApiClient() as enpi_client:
162                template = enpi_client.enrichment_api.get_template(EnrichmentTemplateId(24))
163            ```
164        """
165        enrichment_api_instance = openapi_client.EnrichmentApi(self._inner_api_client)
166
167        with ApiErrorContext():
168            data = enrichment_api_instance.get_enrichment_template(str(enrichment_template_id))
169
170        return EnrichmentTemplate.from_raw(data.template)
171
172    def delete_template(self, enrichment_template_id: EnrichmentTemplateId) -> None:
173        """Delete a Enrichment Template by its ID.
174
175        Args:
176            enrichment_template_id (enpi_api.l2.types.enrichment.EnrichmentTemplateId): ID of the deleted Enrichment template.
177
178        Raises:
179            enpi_api.l2.types.api_error.ApiError: If API request fails.
180
181        Example:
182            ```python
183            with EnpiApiClient() as enpi_client:
184                template = enpi_client.enrichment_api.delete_template(EnrichmentTemplateId(24))
185            ```
186        """
187        enrichment_api_instance = openapi_client.EnrichmentApi(self._inner_api_client)
188
189        with ApiErrorContext():
190            enrichment_api_instance.delete_enrichment_template(str(enrichment_template_id))
191
192    def create_template(
193        self,
194        name: str,
195        operations: list[EnrichmentTemplateOperation],
196    ) -> EnrichmentTemplate:
197        """Create a new Enrichment Template.
198
199        Args:
200            name (str): Enrichment template name.
201            operations (list[enpi_api.l2.types.enrichment.EnrichmentTemplateOperation]): Configs for the template's operations.
202
203        Returns:
204            enpi_api.l2.types.enrichment.EnrichmentTemplate: A newly created Enrichment template.
205
206        Raises:
207            enpi_api.l2.types.api_error.ApiError: If API request fails.
208
209        Example:
210            ```python
211            with EnpiApiClient() as enpi_client:
212                operations=[
213                    EnrichmentTemplateUnionOperation(
214                        name="Counter",
215                    ),
216                    EnrichmentTemplateIntersectionOperation(
217                        name="Target",
218                    ),
219                    EnrichmentTemplateDifferenceOperation(
220                        name="Result",
221                        input_operations=EnrichmentTemplateDifferenceInputs(
222                            remove_operation="Counter",
223                            from_operation="Target",
224                        ),
225                        annotations=[
226                            EnrichmentTemplateFoldChangeAnnotation(name="Target FC"),
227                        ],
228                    ),
229                ],
230                template = enpi_client.enrichment_api.create_template(name="my new template", operations=operations)
231            ```
232        """
233        enrichment_api_instance = openapi_client.EnrichmentApi(self._inner_api_client)
234
235        payload = openapi_client.NewEnrichmentTemplate(name=name, saved=True, operations=[transform_operation(op) for op in operations])
236
237        with ApiErrorContext():
238            data = enrichment_api_instance.create_enrichment_template(payload)
239
240        return EnrichmentTemplate.from_raw(data.template)
241
242    def start(
243        self,
244        name: str,
245        enrichment_template: EnrichmentTemplate,
246        cluster_run_id: ClusterRunId,
247        inputs: list[EnrichmentWorkInput],
248    ) -> Execution[EnrichmentRun]:
249        """Start a new Enrichment run.
250
251        Args:
252            name (str):
253                Enrichment run name.
254            enrichment_template (enpi_api.l2.types.enrichment.EnrichmentTemplate):
255                Enrichment template ID.
256            cluster_run_id (enpi_api.l2.types.cluster.ClusterRunId):
257                Cluster run ID.
258            inputs (list[enpi_api.l2.types.enrichment.EnrichmentWorkInput]):
259                Configs for the template's operations.
260
261        Returns:
262            enpi_api.l2.types.execution.Execution[enpi_api.l2.types.enrichment.EnrichmentRun]: An awaitable that returns the new Enrichment Run.
263
264        Raises:
265            enpi_api.l2.types.api_error.ApiError: If API request fails.
266
267        Example:
268            Assuming that those are the collections meant for Enrichment run:
269                - Counter collection IDs: 1, 2
270                - Target collection IDs: 3, 4
271
272            ```python
273            with EnpiApiClient() as enpi_client:
274                enrichment_run = client.enrichment_api.start(
275                    name="Test Run",
276                    enrichment_template=template,
277                    cluster_run_id=cluster_run.id,
278                    inputs=[
279                        UnionOperationInput(
280                            name="Counter",
281                            input_collections=[CollectionSelector(value=CollectionId(1)),
282                                CollectionSelector(value=CollectionId(2))],
283                        ),
284                        IntersectionOperationInput(
285                            name="Target",
286                            input_collections=[CollectionSelector(value=CollectionId(3)),
287                                CollectionSelector(value=CollectionId(4))
288                            ],
289                        ),
290                        FoldChangeInput(
291                            name="Target FC",
292                            operation_name="Result"
293                            input_collections=FoldChangeInputCollections(
294                                from_collection=CollectionSelector(value=CollectionId(3)),
295                                to_collection=CollectionSelector(value=CollectionId(4)),
296                            ),
297                        ),
298                    ],
299                ).wait()
300            ```
301        """
302        enrichment_api_instance = openapi_client.EnrichmentApi(self._inner_api_client)
303
304        payload = openapi_client.EnrichmentWork(
305            name=name,
306            template=openapi_client.EnrichmentTemplateIdVersion(id=str(enrichment_template.id), version=int(enrichment_template.version)),
307            cluster_run_id=cluster_run_id,
308            inputs=[transform_operation_input(x) for x in inputs],
309        )
310
311        with ApiErrorContext():
312            data = enrichment_api_instance.start_enrichment_run(payload)
313            assert data.workflow_execution_id is not None
314
315            workflow_execution_id = WorkflowExecutionId(int(data.workflow_execution_id))
316
317            def on_complete(task_id: WorkflowExecutionTaskId, task_state: TaskState) -> EnrichmentRun:
318                enrichment_run = self.get_run_by_task_id(task_id)
319
320                logger.success(f"Enrichment run with task ID: {data.workflow_execution_id} has successfully finished.")
321
322                return enrichment_run
323
324            waitable = WorkflowExecutionTaskWaitable[EnrichmentRun](
325                workflow_execution_id=workflow_execution_id, task_template_name=WorkflowTaskTemplateName.ENPI_APP_ENRICHMENT, on_complete=on_complete
326            )
327            return Execution(wait=waitable.wait_and_return_result, check_execution_state=waitable.check_execution_state)
328
329    def export_as_tsv(
330        self,
331        enrichment_run_id: EnrichmentRunId,
332        operation: str,
333        mode: EnrichmentExportMode,
334        tag_ids: list[TagId],
335        limit: int | None = None,
336        output_directory: str | Path | None = None,
337    ) -> Execution[Path]:
338        """Run an export of a Enrichment operation result and download the TSV file.
339
340        Args:
341            enrichment_run_id (enpi_api.l2.types.enrichment.EnrichmentRunId): Enrichment run ID.
342            operation (str): Name of the operation to export.
343            mode (enpi_api.l2.types.enrichment.EnrichmentExportMode): Mode in which export will be run.
344            tag_ids (list[enpi_api.l2.types.tag.TagId]): List of tags to be included in the export.
345            limit (int | None):
346                If specified, the export will contain only the first N clusters, N being the value passed as this param.
347            output_directory (str | Path | None): The directory where to download the TSV file. Defaults to a
348              unique temporary directory.
349
350        Returns:
351            enpi_api.l2.types.execution.Execution[Path]: An awaitable that returns the path to the downloaded TSV file containing the sequences.
352
353        Raises:
354            enpi_api.l2.types.api_error.ApiError: If API request fails.
355
356        Example:
357            ```python
358            with EnpiApiClient() as enpi_client:
359                tsv_path = enpi_client.enrichment_api.export_as_tsv(
360                    enrichment_run_id=EnrichmentRunId(42),
361                    operation="Result",
362                    mode=EnrichmentExportMode.REPRESENTATIVES,
363                    tag_ids=[SequenceTags.Cdr3AminoAcids, SequenceTags.Chain],
364                    limit=50,
365                    output_directory="/results",
366                ).wait()
367            ```
368        """
369
370        # Ensure that the directory exists
371        if output_directory is None:
372            output_directory = unique_temp_dir()
373
374        enrichment_api_instance = openapi_client.EnrichmentApi(self._inner_api_client)
375
376        payload = openapi_client.EnrichmentExportPayload(
377            operation_name=operation,
378            enrichment_run_id=enrichment_run_id,
379            mode=mode,
380            tag_ids=[int(id) for id in tag_ids],
381            limit=limit,
382        )
383
384        with ApiErrorContext():
385            response = enrichment_api_instance.export_enrichment_results(payload)
386            assert response.workflow_execution_id is not None
387
388            workflow_execution_id = WorkflowExecutionId(response.workflow_execution_id)
389
390            def on_complete(task_id: WorkflowExecutionTaskId, task_state: TaskState) -> Path:
391                file_api = FileApi(self._inner_api_client, self._log_level)
392                zip_path = file_api.download_export_by_workflow_execution_task_id(task_id=task_id, output_directory=output_directory)
393
394                with zipfile.ZipFile(zip_path, "r") as archive:
395                    names = archive.namelist()
396                    archive.extractall(output_directory)
397                paths = [Path(os.path.join(output_directory, name)) for name in names]
398
399                if len(paths) > 1:
400                    logger.warning(f"More than 1 file encountered in the zipped export: {','.join([str(path) for path in paths])}, only returning first file")
401                logger.success(f"Enrichment results export with task ID: {task_id} has successfully finished.")
402
403                return paths[0]
404
405            waitable = WorkflowExecutionTaskWaitable[Path](
406                workflow_execution_id=workflow_execution_id, on_complete=on_complete, task_template_name=WorkflowTaskTemplateName.ENPI_APP_ENRICHMENT_EXPORT
407            )
408            return Execution(wait=waitable.wait_and_return_result, check_execution_state=waitable.check_execution_state)
409
410    def export_as_df(
411        self,
412        enrichment_run_id: EnrichmentRunId,
413        operation: str,
414        mode: EnrichmentExportMode,
415        tag_ids: list[TagId],
416        limit: int | None = None,
417    ) -> Execution[pd.DataFrame]:
418        """Runs an Export of a Enrichment Operation results and loads it as a pandas DataFrame.
419
420        Args:
421            enrichment_run_id (enpi_api.l2.types.enrichment.EnrichmentRunId): Enrichment run ID.
422            operation (str): Name of the operation to export.
423            mode (enpi_api.l2.types.enrichment.EnrichmentExportMode): Mode in which export will be run.
424            tag_ids (list[enpi_api.l2.types.tag.TagId]): List of tags to be included in the export.
425            limit (int | None):
426                If specified, the export will contain only the first N clusters, N being the value passed as this param.
427
428        Returns:
429            enpi_api.l2.types.execution.Execution[Path]: An awaitable that returns the path to the downloaded TSV file containing the sequences.
430
431        Raises:
432            enpi_api.l2.types.api_error.ApiError: If API request fails.
433
434        Example:
435            ```python
436            with EnpiApiClient() as enpi_client:
437                tsv_path = enpi_client.enrichment_api.export_as_df(
438                    enrichment_run_id=EnrichmentRunId(42),
439                    operation="Result",
440                    mode=EnrichmentExportMode.REPRESENTATIVES,
441                    tag_ids=[SequenceTags.Cdr3AminoAcids, SequenceTags.Chain],
442                    limit=50
443                ).wait()
444            ```
445        """
446
447        export_tsv = self.export_as_tsv(enrichment_run_id, operation, mode, tag_ids, limit)
448
449        def wait() -> pd.DataFrame:
450            file_path = export_tsv.wait()
451            return pd.read_csv(file_path, delimiter="\t")
452
453        return Execution(wait=wait, check_execution_state=export_tsv.check_execution_state)

class EnrichmentApi: View Source

 34class EnrichmentApi:
 35    _inner_api_client: openapi_client.ApiClient
 36    _log_level: LogLevel
 37
 38    def __init__(self, inner_api_client: openapi_client.ApiClient, log_level: LogLevel):
 39        """@private"""
 40        self._inner_api_client = inner_api_client
 41        self._log_level = log_level
 42
 43    def get_runs(
 44        self,
 45    ) -> list[EnrichmentRun]:
 46        """Get all successful Enrichment Runs.
 47
 48        Returns:
 49            list[enpi_api.l2.types.enrichment.EnrichmentRun]: List of Enrichment Runs.
 50
 51        Raises:
 52            enpi_api.l2.types.api_error.ApiError: If API request fails.
 53
 54        Example:
 55            ```python
 56            with EnpiApiClient() as enpi_client:
 57                enrichment_runs = enpi_client.enrichment_api.get_runs()
 58            ```
 59        """
 60        enrichment_api_instance = openapi_client.EnrichmentApi(self._inner_api_client)
 61
 62        with ApiErrorContext():
 63            data = enrichment_api_instance.get_enrichment_runs()
 64
 65        return [EnrichmentRun.from_raw(cr) for cr in data.runs]
 66
 67    def get_run(self, enrichment_run_id: EnrichmentRunId) -> EnrichmentRun:
 68        """Get a single Enrichment Run by its ID.
 69
 70        Args:
 71            enrichment_run_id (enpi_api.l2.types.enrichment.EnrichmentRunId): ID of the Enrichment run to get.
 72
 73        Returns:
 74            enpi_api.l2.types.enrichment.EnrichmentRun: A successful Enrichment Run.
 75
 76        Raises:
 77            enpi_api.l2.types.api_error.ApiError: If API request fails.
 78
 79        Example:
 80            ```python
 81            with EnpiApiClient() as enpi_client:
 82                enrichment_run = enpi_client.enrichment_api.get_run(EnrichmentRunId(123))
 83            ```
 84        """
 85        enrichment_api_instance = openapi_client.EnrichmentApi(self._inner_api_client)
 86
 87        with ApiErrorContext():
 88            data = enrichment_api_instance.get_enrichment_run(int(enrichment_run_id))
 89
 90        return EnrichmentRun.from_raw(data.run)
 91
 92    def get_run_by_task_id(self, task_id: WorkflowExecutionTaskId) -> EnrichmentRun:
 93        """Get a single Enrichment Run by its task ID.
 94
 95        Args:
 96            task_id (enpi_api.l2.types.task.TaskId): ID of a task linked to a successful Enrichment Run.
 97
 98        Returns:
 99            enpi_api.l2.types.enrichment.EnrichmentRun: Successful Enrichment Run linked to the provided task ID.
100
101        Raises:
102            enpi_api.l2.types.api_error.ApiError: If API request fails.
103
104        Example:
105            Fetch a Enrichment Run
106
107            ```python
108            with EnpiApiClient() as enpi_client:
109                enrichment_run = enpi_client.enrichment_api.get_run_by_task_id(TaskId(1234))
110            ```
111        """
112        enrichment_api_instance = openapi_client.EnrichmentApi(self._inner_api_client)
113
114        with ApiErrorContext():
115            data = enrichment_api_instance.get_enrichment_run_by_task_id(task_id)
116
117        return EnrichmentRun.from_raw(data.run)
118
119    def get_templates(self) -> list[SimplifiedEnrichmentTemplate]:
120        """Get all available Enrichment templates.
121
122        Returns:
123            list[enpi_api.l2.types.enrichment.SimplifiedEnrichmentTemplate]: Available Enrichment templates.
124
125        Raises:
126            enpi_api.l2.types.api_error.ApiError: If API request fails.
127
128        Example:
129            ```python
130            with EnpiApiClient() as enpi_client:
131                templates = enpi_client.enrichment_api.get_templates()
132            ```
133        """
134        enrichment_api_instance = openapi_client.EnrichmentApi(self._inner_api_client)
135
136        with ApiErrorContext():
137            data = enrichment_api_instance.get_enrichment_templates()
138
139        return [
140            SimplifiedEnrichmentTemplate(
141                id=EnrichmentTemplateId(d.id),
142                name=d.name,
143                created_at=d.created_at,
144            )
145            for d in data.templates
146        ]
147
148    def get_template(self, enrichment_template_id: EnrichmentTemplateId) -> EnrichmentTemplate:
149        """Get a Enrichment template by its ID.
150
151        Args:
152            enrichment_template_id (enpi_api.l2.types.enrichment.EnrichmentTemplateId): ID of a Enrichment template to get.
153
154        Returns:
155            enpi_api.l2.types.enrichment.EnrichmentTemplate: A Enrichment template matching the provided ID.
156
157        Raises:
158            enpi_api.l2.types.api_error.ApiError: If API request fails.
159
160        Example:
161            ```python
162            with EnpiApiClient() as enpi_client:
163                template = enpi_client.enrichment_api.get_template(EnrichmentTemplateId(24))
164            ```
165        """
166        enrichment_api_instance = openapi_client.EnrichmentApi(self._inner_api_client)
167
168        with ApiErrorContext():
169            data = enrichment_api_instance.get_enrichment_template(str(enrichment_template_id))
170
171        return EnrichmentTemplate.from_raw(data.template)
172
173    def delete_template(self, enrichment_template_id: EnrichmentTemplateId) -> None:
174        """Delete a Enrichment Template by its ID.
175
176        Args:
177            enrichment_template_id (enpi_api.l2.types.enrichment.EnrichmentTemplateId): ID of the deleted Enrichment template.
178
179        Raises:
180            enpi_api.l2.types.api_error.ApiError: If API request fails.
181
182        Example:
183            ```python
184            with EnpiApiClient() as enpi_client:
185                template = enpi_client.enrichment_api.delete_template(EnrichmentTemplateId(24))
186            ```
187        """
188        enrichment_api_instance = openapi_client.EnrichmentApi(self._inner_api_client)
189
190        with ApiErrorContext():
191            enrichment_api_instance.delete_enrichment_template(str(enrichment_template_id))
192
193    def create_template(
194        self,
195        name: str,
196        operations: list[EnrichmentTemplateOperation],
197    ) -> EnrichmentTemplate:
198        """Create a new Enrichment Template.
199
200        Args:
201            name (str): Enrichment template name.
202            operations (list[enpi_api.l2.types.enrichment.EnrichmentTemplateOperation]): Configs for the template's operations.
203
204        Returns:
205            enpi_api.l2.types.enrichment.EnrichmentTemplate: A newly created Enrichment template.
206
207        Raises:
208            enpi_api.l2.types.api_error.ApiError: If API request fails.
209
210        Example:
211            ```python
212            with EnpiApiClient() as enpi_client:
213                operations=[
214                    EnrichmentTemplateUnionOperation(
215                        name="Counter",
216                    ),
217                    EnrichmentTemplateIntersectionOperation(
218                        name="Target",
219                    ),
220                    EnrichmentTemplateDifferenceOperation(
221                        name="Result",
222                        input_operations=EnrichmentTemplateDifferenceInputs(
223                            remove_operation="Counter",
224                            from_operation="Target",
225                        ),
226                        annotations=[
227                            EnrichmentTemplateFoldChangeAnnotation(name="Target FC"),
228                        ],
229                    ),
230                ],
231                template = enpi_client.enrichment_api.create_template(name="my new template", operations=operations)
232            ```
233        """
234        enrichment_api_instance = openapi_client.EnrichmentApi(self._inner_api_client)
235
236        payload = openapi_client.NewEnrichmentTemplate(name=name, saved=True, operations=[transform_operation(op) for op in operations])
237
238        with ApiErrorContext():
239            data = enrichment_api_instance.create_enrichment_template(payload)
240
241        return EnrichmentTemplate.from_raw(data.template)
242
243    def start(
244        self,
245        name: str,
246        enrichment_template: EnrichmentTemplate,
247        cluster_run_id: ClusterRunId,
248        inputs: list[EnrichmentWorkInput],
249    ) -> Execution[EnrichmentRun]:
250        """Start a new Enrichment run.
251
252        Args:
253            name (str):
254                Enrichment run name.
255            enrichment_template (enpi_api.l2.types.enrichment.EnrichmentTemplate):
256                Enrichment template ID.
257            cluster_run_id (enpi_api.l2.types.cluster.ClusterRunId):
258                Cluster run ID.
259            inputs (list[enpi_api.l2.types.enrichment.EnrichmentWorkInput]):
260                Configs for the template's operations.
261
262        Returns:
263            enpi_api.l2.types.execution.Execution[enpi_api.l2.types.enrichment.EnrichmentRun]: An awaitable that returns the new Enrichment Run.
264
265        Raises:
266            enpi_api.l2.types.api_error.ApiError: If API request fails.
267
268        Example:
269            Assuming that those are the collections meant for Enrichment run:
270                - Counter collection IDs: 1, 2
271                - Target collection IDs: 3, 4
272
273            ```python
274            with EnpiApiClient() as enpi_client:
275                enrichment_run = client.enrichment_api.start(
276                    name="Test Run",
277                    enrichment_template=template,
278                    cluster_run_id=cluster_run.id,
279                    inputs=[
280                        UnionOperationInput(
281                            name="Counter",
282                            input_collections=[CollectionSelector(value=CollectionId(1)),
283                                CollectionSelector(value=CollectionId(2))],
284                        ),
285                        IntersectionOperationInput(
286                            name="Target",
287                            input_collections=[CollectionSelector(value=CollectionId(3)),
288                                CollectionSelector(value=CollectionId(4))
289                            ],
290                        ),
291                        FoldChangeInput(
292                            name="Target FC",
293                            operation_name="Result"
294                            input_collections=FoldChangeInputCollections(
295                                from_collection=CollectionSelector(value=CollectionId(3)),
296                                to_collection=CollectionSelector(value=CollectionId(4)),
297                            ),
298                        ),
299                    ],
300                ).wait()
301            ```
302        """
303        enrichment_api_instance = openapi_client.EnrichmentApi(self._inner_api_client)
304
305        payload = openapi_client.EnrichmentWork(
306            name=name,
307            template=openapi_client.EnrichmentTemplateIdVersion(id=str(enrichment_template.id), version=int(enrichment_template.version)),
308            cluster_run_id=cluster_run_id,
309            inputs=[transform_operation_input(x) for x in inputs],
310        )
311
312        with ApiErrorContext():
313            data = enrichment_api_instance.start_enrichment_run(payload)
314            assert data.workflow_execution_id is not None
315
316            workflow_execution_id = WorkflowExecutionId(int(data.workflow_execution_id))
317
318            def on_complete(task_id: WorkflowExecutionTaskId, task_state: TaskState) -> EnrichmentRun:
319                enrichment_run = self.get_run_by_task_id(task_id)
320
321                logger.success(f"Enrichment run with task ID: {data.workflow_execution_id} has successfully finished.")
322
323                return enrichment_run
324
325            waitable = WorkflowExecutionTaskWaitable[EnrichmentRun](
326                workflow_execution_id=workflow_execution_id, task_template_name=WorkflowTaskTemplateName.ENPI_APP_ENRICHMENT, on_complete=on_complete
327            )
328            return Execution(wait=waitable.wait_and_return_result, check_execution_state=waitable.check_execution_state)
329
330    def export_as_tsv(
331        self,
332        enrichment_run_id: EnrichmentRunId,
333        operation: str,
334        mode: EnrichmentExportMode,
335        tag_ids: list[TagId],
336        limit: int | None = None,
337        output_directory: str | Path | None = None,
338    ) -> Execution[Path]:
339        """Run an export of a Enrichment operation result and download the TSV file.
340
341        Args:
342            enrichment_run_id (enpi_api.l2.types.enrichment.EnrichmentRunId): Enrichment run ID.
343            operation (str): Name of the operation to export.
344            mode (enpi_api.l2.types.enrichment.EnrichmentExportMode): Mode in which export will be run.
345            tag_ids (list[enpi_api.l2.types.tag.TagId]): List of tags to be included in the export.
346            limit (int | None):
347                If specified, the export will contain only the first N clusters, N being the value passed as this param.
348            output_directory (str | Path | None): The directory where to download the TSV file. Defaults to a
349              unique temporary directory.
350
351        Returns:
352            enpi_api.l2.types.execution.Execution[Path]: An awaitable that returns the path to the downloaded TSV file containing the sequences.
353
354        Raises:
355            enpi_api.l2.types.api_error.ApiError: If API request fails.
356
357        Example:
358            ```python
359            with EnpiApiClient() as enpi_client:
360                tsv_path = enpi_client.enrichment_api.export_as_tsv(
361                    enrichment_run_id=EnrichmentRunId(42),
362                    operation="Result",
363                    mode=EnrichmentExportMode.REPRESENTATIVES,
364                    tag_ids=[SequenceTags.Cdr3AminoAcids, SequenceTags.Chain],
365                    limit=50,
366                    output_directory="/results",
367                ).wait()
368            ```
369        """
370
371        # Ensure that the directory exists
372        if output_directory is None:
373            output_directory = unique_temp_dir()
374
375        enrichment_api_instance = openapi_client.EnrichmentApi(self._inner_api_client)
376
377        payload = openapi_client.EnrichmentExportPayload(
378            operation_name=operation,
379            enrichment_run_id=enrichment_run_id,
380            mode=mode,
381            tag_ids=[int(id) for id in tag_ids],
382            limit=limit,
383        )
384
385        with ApiErrorContext():
386            response = enrichment_api_instance.export_enrichment_results(payload)
387            assert response.workflow_execution_id is not None
388
389            workflow_execution_id = WorkflowExecutionId(response.workflow_execution_id)
390
391            def on_complete(task_id: WorkflowExecutionTaskId, task_state: TaskState) -> Path:
392                file_api = FileApi(self._inner_api_client, self._log_level)
393                zip_path = file_api.download_export_by_workflow_execution_task_id(task_id=task_id, output_directory=output_directory)
394
395                with zipfile.ZipFile(zip_path, "r") as archive:
396                    names = archive.namelist()
397                    archive.extractall(output_directory)
398                paths = [Path(os.path.join(output_directory, name)) for name in names]
399
400                if len(paths) > 1:
401                    logger.warning(f"More than 1 file encountered in the zipped export: {','.join([str(path) for path in paths])}, only returning first file")
402                logger.success(f"Enrichment results export with task ID: {task_id} has successfully finished.")
403
404                return paths[0]
405
406            waitable = WorkflowExecutionTaskWaitable[Path](
407                workflow_execution_id=workflow_execution_id, on_complete=on_complete, task_template_name=WorkflowTaskTemplateName.ENPI_APP_ENRICHMENT_EXPORT
408            )
409            return Execution(wait=waitable.wait_and_return_result, check_execution_state=waitable.check_execution_state)
410
411    def export_as_df(
412        self,
413        enrichment_run_id: EnrichmentRunId,
414        operation: str,
415        mode: EnrichmentExportMode,
416        tag_ids: list[TagId],
417        limit: int | None = None,
418    ) -> Execution[pd.DataFrame]:
419        """Runs an Export of a Enrichment Operation results and loads it as a pandas DataFrame.
420
421        Args:
422            enrichment_run_id (enpi_api.l2.types.enrichment.EnrichmentRunId): Enrichment run ID.
423            operation (str): Name of the operation to export.
424            mode (enpi_api.l2.types.enrichment.EnrichmentExportMode): Mode in which export will be run.
425            tag_ids (list[enpi_api.l2.types.tag.TagId]): List of tags to be included in the export.
426            limit (int | None):
427                If specified, the export will contain only the first N clusters, N being the value passed as this param.
428
429        Returns:
430            enpi_api.l2.types.execution.Execution[Path]: An awaitable that returns the path to the downloaded TSV file containing the sequences.
431
432        Raises:
433            enpi_api.l2.types.api_error.ApiError: If API request fails.
434
435        Example:
436            ```python
437            with EnpiApiClient() as enpi_client:
438                tsv_path = enpi_client.enrichment_api.export_as_df(
439                    enrichment_run_id=EnrichmentRunId(42),
440                    operation="Result",
441                    mode=EnrichmentExportMode.REPRESENTATIVES,
442                    tag_ids=[SequenceTags.Cdr3AminoAcids, SequenceTags.Chain],
443                    limit=50
444                ).wait()
445            ```
446        """
447
448        export_tsv = self.export_as_tsv(enrichment_run_id, operation, mode, tag_ids, limit)
449
450        def wait() -> pd.DataFrame:
451            file_path = export_tsv.wait()
452            return pd.read_csv(file_path, delimiter="\t")
453
454        return Execution(wait=wait, check_execution_state=export_tsv.check_execution_state)

def get_runs(self) -> list[enpi_api.l2.types.enrichment.EnrichmentRun]: View Source

43    def get_runs(
44        self,
45    ) -> list[EnrichmentRun]:
46        """Get all successful Enrichment Runs.
47
48        Returns:
49            list[enpi_api.l2.types.enrichment.EnrichmentRun]: List of Enrichment Runs.
50
51        Raises:
52            enpi_api.l2.types.api_error.ApiError: If API request fails.
53
54        Example:
55            ```python
56            with EnpiApiClient() as enpi_client:
57                enrichment_runs = enpi_client.enrichment_api.get_runs()
58            ```
59        """
60        enrichment_api_instance = openapi_client.EnrichmentApi(self._inner_api_client)
61
62        with ApiErrorContext():
63            data = enrichment_api_instance.get_enrichment_runs()
64
65        return [EnrichmentRun.from_raw(cr) for cr in data.runs]

Get all successful Enrichment Runs.

Returns:

list[enpi_api.l2.types.enrichment.EnrichmentRun]: List of Enrichment Runs.

Raises:

enpi_api.l2.types.api_error.ApiError: If API request fails.

Example:

with EnpiApiClient() as enpi_client:
    enrichment_runs = enpi_client.enrichment_api.get_runs()

def get_run( self, enrichment_run_id: enpi_api.l2.types.enrichment.EnrichmentRunId) -> enpi_api.l2.types.enrichment.EnrichmentRun: View Source

67    def get_run(self, enrichment_run_id: EnrichmentRunId) -> EnrichmentRun:
68        """Get a single Enrichment Run by its ID.
69
70        Args:
71            enrichment_run_id (enpi_api.l2.types.enrichment.EnrichmentRunId): ID of the Enrichment run to get.
72
73        Returns:
74            enpi_api.l2.types.enrichment.EnrichmentRun: A successful Enrichment Run.
75
76        Raises:
77            enpi_api.l2.types.api_error.ApiError: If API request fails.
78
79        Example:
80            ```python
81            with EnpiApiClient() as enpi_client:
82                enrichment_run = enpi_client.enrichment_api.get_run(EnrichmentRunId(123))
83            ```
84        """
85        enrichment_api_instance = openapi_client.EnrichmentApi(self._inner_api_client)
86
87        with ApiErrorContext():
88            data = enrichment_api_instance.get_enrichment_run(int(enrichment_run_id))
89
90        return EnrichmentRun.from_raw(data.run)

Get a single Enrichment Run by its ID.

Arguments:

enrichment_run_id (enpi_api.l2.types.enrichment.EnrichmentRunId): ID of the Enrichment run to get.

Returns:

enpi_api.l2.types.enrichment.EnrichmentRun: A successful Enrichment Run.

Raises:

enpi_api.l2.types.api_error.ApiError: If API request fails.

Example:

with EnpiApiClient() as enpi_client:
    enrichment_run = enpi_client.enrichment_api.get_run(EnrichmentRunId(123))

def get_run_by_task_id( self, task_id: enpi_api.l2.types.workflow.WorkflowExecutionTaskId) -> enpi_api.l2.types.enrichment.EnrichmentRun: View Source

 92    def get_run_by_task_id(self, task_id: WorkflowExecutionTaskId) -> EnrichmentRun:
 93        """Get a single Enrichment Run by its task ID.
 94
 95        Args:
 96            task_id (enpi_api.l2.types.task.TaskId): ID of a task linked to a successful Enrichment Run.
 97
 98        Returns:
 99            enpi_api.l2.types.enrichment.EnrichmentRun: Successful Enrichment Run linked to the provided task ID.
100
101        Raises:
102            enpi_api.l2.types.api_error.ApiError: If API request fails.
103
104        Example:
105            Fetch a Enrichment Run
106
107            ```python
108            with EnpiApiClient() as enpi_client:
109                enrichment_run = enpi_client.enrichment_api.get_run_by_task_id(TaskId(1234))
110            ```
111        """
112        enrichment_api_instance = openapi_client.EnrichmentApi(self._inner_api_client)
113
114        with ApiErrorContext():
115            data = enrichment_api_instance.get_enrichment_run_by_task_id(task_id)
116
117        return EnrichmentRun.from_raw(data.run)

Get a single Enrichment Run by its task ID.

Arguments:

task_id (enpi_api.l2.types.task.TaskId): ID of a task linked to a successful Enrichment Run.

Returns:

enpi_api.l2.types.enrichment.EnrichmentRun: Successful Enrichment Run linked to the provided task ID.

Raises:

enpi_api.l2.types.api_error.ApiError: If API request fails.

Example:

Fetch a Enrichment Run

with EnpiApiClient() as enpi_client:
    enrichment_run = enpi_client.enrichment_api.get_run_by_task_id(TaskId(1234))

def get_templates(self) -> list[enpi_api.l2.types.enrichment.SimplifiedEnrichmentTemplate]: View Source

119    def get_templates(self) -> list[SimplifiedEnrichmentTemplate]:
120        """Get all available Enrichment templates.
121
122        Returns:
123            list[enpi_api.l2.types.enrichment.SimplifiedEnrichmentTemplate]: Available Enrichment templates.
124
125        Raises:
126            enpi_api.l2.types.api_error.ApiError: If API request fails.
127
128        Example:
129            ```python
130            with EnpiApiClient() as enpi_client:
131                templates = enpi_client.enrichment_api.get_templates()
132            ```
133        """
134        enrichment_api_instance = openapi_client.EnrichmentApi(self._inner_api_client)
135
136        with ApiErrorContext():
137            data = enrichment_api_instance.get_enrichment_templates()
138
139        return [
140            SimplifiedEnrichmentTemplate(
141                id=EnrichmentTemplateId(d.id),
142                name=d.name,
143                created_at=d.created_at,
144            )
145            for d in data.templates
146        ]

Get all available Enrichment templates.

Returns:

list[enpi_api.l2.types.enrichment.SimplifiedEnrichmentTemplate]: Available Enrichment templates.

Raises:

enpi_api.l2.types.api_error.ApiError: If API request fails.

Example:

with EnpiApiClient() as enpi_client:
    templates = enpi_client.enrichment_api.get_templates()

def get_template( self, enrichment_template_id: enpi_api.l2.types.enrichment.EnrichmentTemplateId) -> enpi_api.l2.types.enrichment.EnrichmentTemplate: View Source

148    def get_template(self, enrichment_template_id: EnrichmentTemplateId) -> EnrichmentTemplate:
149        """Get a Enrichment template by its ID.
150
151        Args:
152            enrichment_template_id (enpi_api.l2.types.enrichment.EnrichmentTemplateId): ID of a Enrichment template to get.
153
154        Returns:
155            enpi_api.l2.types.enrichment.EnrichmentTemplate: A Enrichment template matching the provided ID.
156
157        Raises:
158            enpi_api.l2.types.api_error.ApiError: If API request fails.
159
160        Example:
161            ```python
162            with EnpiApiClient() as enpi_client:
163                template = enpi_client.enrichment_api.get_template(EnrichmentTemplateId(24))
164            ```
165        """
166        enrichment_api_instance = openapi_client.EnrichmentApi(self._inner_api_client)
167
168        with ApiErrorContext():
169            data = enrichment_api_instance.get_enrichment_template(str(enrichment_template_id))
170
171        return EnrichmentTemplate.from_raw(data.template)

Get a Enrichment template by its ID.

Arguments:

enrichment_template_id (enpi_api.l2.types.enrichment.EnrichmentTemplateId): ID of a Enrichment template to get.

Returns:

enpi_api.l2.types.enrichment.EnrichmentTemplate: A Enrichment template matching the provided ID.

Raises:

enpi_api.l2.types.api_error.ApiError: If API request fails.

Example:

with EnpiApiClient() as enpi_client:
    template = enpi_client.enrichment_api.get_template(EnrichmentTemplateId(24))

def delete_template( self, enrichment_template_id: enpi_api.l2.types.enrichment.EnrichmentTemplateId) -> None: View Source

173    def delete_template(self, enrichment_template_id: EnrichmentTemplateId) -> None:
174        """Delete a Enrichment Template by its ID.
175
176        Args:
177            enrichment_template_id (enpi_api.l2.types.enrichment.EnrichmentTemplateId): ID of the deleted Enrichment template.
178
179        Raises:
180            enpi_api.l2.types.api_error.ApiError: If API request fails.
181
182        Example:
183            ```python
184            with EnpiApiClient() as enpi_client:
185                template = enpi_client.enrichment_api.delete_template(EnrichmentTemplateId(24))
186            ```
187        """
188        enrichment_api_instance = openapi_client.EnrichmentApi(self._inner_api_client)
189
190        with ApiErrorContext():
191            enrichment_api_instance.delete_enrichment_template(str(enrichment_template_id))

Delete a Enrichment Template by its ID.

Arguments:

enrichment_template_id (enpi_api.l2.types.enrichment.EnrichmentTemplateId): ID of the deleted Enrichment template.

Raises:

enpi_api.l2.types.api_error.ApiError: If API request fails.

Example:

with EnpiApiClient() as enpi_client:
    template = enpi_client.enrichment_api.delete_template(EnrichmentTemplateId(24))

def create_template( self, name: str, operations: list[enpi_api.l2.types.enrichment.EnrichmentTemplateUnionOperation | enpi_api.l2.types.enrichment.EnrichmentTemplateIntersectionOperation | enpi_api.l2.types.enrichment.EnrichmentTemplateDifferenceOperation]) -> enpi_api.l2.types.enrichment.EnrichmentTemplate: View Source

193    def create_template(
194        self,
195        name: str,
196        operations: list[EnrichmentTemplateOperation],
197    ) -> EnrichmentTemplate:
198        """Create a new Enrichment Template.
199
200        Args:
201            name (str): Enrichment template name.
202            operations (list[enpi_api.l2.types.enrichment.EnrichmentTemplateOperation]): Configs for the template's operations.
203
204        Returns:
205            enpi_api.l2.types.enrichment.EnrichmentTemplate: A newly created Enrichment template.
206
207        Raises:
208            enpi_api.l2.types.api_error.ApiError: If API request fails.
209
210        Example:
211            ```python
212            with EnpiApiClient() as enpi_client:
213                operations=[
214                    EnrichmentTemplateUnionOperation(
215                        name="Counter",
216                    ),
217                    EnrichmentTemplateIntersectionOperation(
218                        name="Target",
219                    ),
220                    EnrichmentTemplateDifferenceOperation(
221                        name="Result",
222                        input_operations=EnrichmentTemplateDifferenceInputs(
223                            remove_operation="Counter",
224                            from_operation="Target",
225                        ),
226                        annotations=[
227                            EnrichmentTemplateFoldChangeAnnotation(name="Target FC"),
228                        ],
229                    ),
230                ],
231                template = enpi_client.enrichment_api.create_template(name="my new template", operations=operations)
232            ```
233        """
234        enrichment_api_instance = openapi_client.EnrichmentApi(self._inner_api_client)
235
236        payload = openapi_client.NewEnrichmentTemplate(name=name, saved=True, operations=[transform_operation(op) for op in operations])
237
238        with ApiErrorContext():
239            data = enrichment_api_instance.create_enrichment_template(payload)
240
241        return EnrichmentTemplate.from_raw(data.template)

Create a new Enrichment Template.

Arguments:

name (str): Enrichment template name.
operations (list[enpi_api.l2.types.enrichment.EnrichmentTemplateOperation]): Configs for the template's operations.

Returns:

enpi_api.l2.types.enrichment.EnrichmentTemplate: A newly created Enrichment template.

Raises:

enpi_api.l2.types.api_error.ApiError: If API request fails.

Example:

with EnpiApiClient() as enpi_client:
    operations=[
        EnrichmentTemplateUnionOperation(
            name="Counter",
        ),
        EnrichmentTemplateIntersectionOperation(
            name="Target",
        ),
        EnrichmentTemplateDifferenceOperation(
            name="Result",
            input_operations=EnrichmentTemplateDifferenceInputs(
                remove_operation="Counter",
                from_operation="Target",
            ),
            annotations=[
                EnrichmentTemplateFoldChangeAnnotation(name="Target FC"),
            ],
        ),
    ],
    template = enpi_client.enrichment_api.create_template(name="my new template", operations=operations)

def start( self, name: str, enrichment_template: enpi_api.l2.types.enrichment.EnrichmentTemplate, cluster_run_id: enpi_api.l2.types.cluster.ClusterRunId, inputs: list[enpi_api.l2.types.enrichment.UnionOperationInput | enpi_api.l2.types.enrichment.IntersectionOperationInput | enpi_api.l2.types.enrichment.DifferenceOperationInput | enpi_api.l2.types.enrichment.FoldChangeInput]) -> enpi_api.l2.types.execution.Execution[EnrichmentRun]: View Source

243    def start(
244        self,
245        name: str,
246        enrichment_template: EnrichmentTemplate,
247        cluster_run_id: ClusterRunId,
248        inputs: list[EnrichmentWorkInput],
249    ) -> Execution[EnrichmentRun]:
250        """Start a new Enrichment run.
251
252        Args:
253            name (str):
254                Enrichment run name.
255            enrichment_template (enpi_api.l2.types.enrichment.EnrichmentTemplate):
256                Enrichment template ID.
257            cluster_run_id (enpi_api.l2.types.cluster.ClusterRunId):
258                Cluster run ID.
259            inputs (list[enpi_api.l2.types.enrichment.EnrichmentWorkInput]):
260                Configs for the template's operations.
261
262        Returns:
263            enpi_api.l2.types.execution.Execution[enpi_api.l2.types.enrichment.EnrichmentRun]: An awaitable that returns the new Enrichment Run.
264
265        Raises:
266            enpi_api.l2.types.api_error.ApiError: If API request fails.
267
268        Example:
269            Assuming that those are the collections meant for Enrichment run:
270                - Counter collection IDs: 1, 2
271                - Target collection IDs: 3, 4
272
273            ```python
274            with EnpiApiClient() as enpi_client:
275                enrichment_run = client.enrichment_api.start(
276                    name="Test Run",
277                    enrichment_template=template,
278                    cluster_run_id=cluster_run.id,
279                    inputs=[
280                        UnionOperationInput(
281                            name="Counter",
282                            input_collections=[CollectionSelector(value=CollectionId(1)),
283                                CollectionSelector(value=CollectionId(2))],
284                        ),
285                        IntersectionOperationInput(
286                            name="Target",
287                            input_collections=[CollectionSelector(value=CollectionId(3)),
288                                CollectionSelector(value=CollectionId(4))
289                            ],
290                        ),
291                        FoldChangeInput(
292                            name="Target FC",
293                            operation_name="Result"
294                            input_collections=FoldChangeInputCollections(
295                                from_collection=CollectionSelector(value=CollectionId(3)),
296                                to_collection=CollectionSelector(value=CollectionId(4)),
297                            ),
298                        ),
299                    ],
300                ).wait()
301            ```
302        """
303        enrichment_api_instance = openapi_client.EnrichmentApi(self._inner_api_client)
304
305        payload = openapi_client.EnrichmentWork(
306            name=name,
307            template=openapi_client.EnrichmentTemplateIdVersion(id=str(enrichment_template.id), version=int(enrichment_template.version)),
308            cluster_run_id=cluster_run_id,
309            inputs=[transform_operation_input(x) for x in inputs],
310        )
311
312        with ApiErrorContext():
313            data = enrichment_api_instance.start_enrichment_run(payload)
314            assert data.workflow_execution_id is not None
315
316            workflow_execution_id = WorkflowExecutionId(int(data.workflow_execution_id))
317
318            def on_complete(task_id: WorkflowExecutionTaskId, task_state: TaskState) -> EnrichmentRun:
319                enrichment_run = self.get_run_by_task_id(task_id)
320
321                logger.success(f"Enrichment run with task ID: {data.workflow_execution_id} has successfully finished.")
322
323                return enrichment_run
324
325            waitable = WorkflowExecutionTaskWaitable[EnrichmentRun](
326                workflow_execution_id=workflow_execution_id, task_template_name=WorkflowTaskTemplateName.ENPI_APP_ENRICHMENT, on_complete=on_complete
327            )
328            return Execution(wait=waitable.wait_and_return_result, check_execution_state=waitable.check_execution_state)

Start a new Enrichment run.

Arguments:

name (str): Enrichment run name.
enrichment_template (enpi_api.l2.types.enrichment.EnrichmentTemplate): Enrichment template ID.
cluster_run_id (enpi_api.l2.types.cluster.ClusterRunId): Cluster run ID.
inputs (list[enpi_api.l2.types.enrichment.EnrichmentWorkInput]): Configs for the template's operations.

Returns:

enpi_api.l2.types.execution.Execution[enpi_api.l2.types.enrichment.EnrichmentRun]: An awaitable that returns the new Enrichment Run.

Raises:

enpi_api.l2.types.api_error.ApiError: If API request fails.

Example:

Assuming that those are the collections meant for Enrichment run: - Counter collection IDs: 1, 2 - Target collection IDs: 3, 4

with EnpiApiClient() as enpi_client:
    enrichment_run = client.enrichment_api.start(
        name="Test Run",
        enrichment_template=template,
        cluster_run_id=cluster_run.id,
        inputs=[
            UnionOperationInput(
                name="Counter",
                input_collections=[CollectionSelector(value=CollectionId(1)),
                    CollectionSelector(value=CollectionId(2))],
            ),
            IntersectionOperationInput(
                name="Target",
                input_collections=[CollectionSelector(value=CollectionId(3)),
                    CollectionSelector(value=CollectionId(4))
                ],
            ),
            FoldChangeInput(
                name="Target FC",
                operation_name="Result"
                input_collections=FoldChangeInputCollections(
                    from_collection=CollectionSelector(value=CollectionId(3)),
                    to_collection=CollectionSelector(value=CollectionId(4)),
                ),
            ),
        ],
    ).wait()

def export_as_tsv( self, enrichment_run_id: enpi_api.l2.types.enrichment.EnrichmentRunId, operation: str, mode: enpi_api.l2.types.enrichment.EnrichmentExportMode, tag_ids: list[enpi_api.l2.types.tag.TagId], limit: int | None = None, output_directory: str | pathlib.Path | None = None) -> enpi_api.l2.types.execution.Execution[Path]: View Source

330    def export_as_tsv(
331        self,
332        enrichment_run_id: EnrichmentRunId,
333        operation: str,
334        mode: EnrichmentExportMode,
335        tag_ids: list[TagId],
336        limit: int | None = None,
337        output_directory: str | Path | None = None,
338    ) -> Execution[Path]:
339        """Run an export of a Enrichment operation result and download the TSV file.
340
341        Args:
342            enrichment_run_id (enpi_api.l2.types.enrichment.EnrichmentRunId): Enrichment run ID.
343            operation (str): Name of the operation to export.
344            mode (enpi_api.l2.types.enrichment.EnrichmentExportMode): Mode in which export will be run.
345            tag_ids (list[enpi_api.l2.types.tag.TagId]): List of tags to be included in the export.
346            limit (int | None):
347                If specified, the export will contain only the first N clusters, N being the value passed as this param.
348            output_directory (str | Path | None): The directory where to download the TSV file. Defaults to a
349              unique temporary directory.
350
351        Returns:
352            enpi_api.l2.types.execution.Execution[Path]: An awaitable that returns the path to the downloaded TSV file containing the sequences.
353
354        Raises:
355            enpi_api.l2.types.api_error.ApiError: If API request fails.
356
357        Example:
358            ```python
359            with EnpiApiClient() as enpi_client:
360                tsv_path = enpi_client.enrichment_api.export_as_tsv(
361                    enrichment_run_id=EnrichmentRunId(42),
362                    operation="Result",
363                    mode=EnrichmentExportMode.REPRESENTATIVES,
364                    tag_ids=[SequenceTags.Cdr3AminoAcids, SequenceTags.Chain],
365                    limit=50,
366                    output_directory="/results",
367                ).wait()
368            ```
369        """
370
371        # Ensure that the directory exists
372        if output_directory is None:
373            output_directory = unique_temp_dir()
374
375        enrichment_api_instance = openapi_client.EnrichmentApi(self._inner_api_client)
376
377        payload = openapi_client.EnrichmentExportPayload(
378            operation_name=operation,
379            enrichment_run_id=enrichment_run_id,
380            mode=mode,
381            tag_ids=[int(id) for id in tag_ids],
382            limit=limit,
383        )
384
385        with ApiErrorContext():
386            response = enrichment_api_instance.export_enrichment_results(payload)
387            assert response.workflow_execution_id is not None
388
389            workflow_execution_id = WorkflowExecutionId(response.workflow_execution_id)
390
391            def on_complete(task_id: WorkflowExecutionTaskId, task_state: TaskState) -> Path:
392                file_api = FileApi(self._inner_api_client, self._log_level)
393                zip_path = file_api.download_export_by_workflow_execution_task_id(task_id=task_id, output_directory=output_directory)
394
395                with zipfile.ZipFile(zip_path, "r") as archive:
396                    names = archive.namelist()
397                    archive.extractall(output_directory)
398                paths = [Path(os.path.join(output_directory, name)) for name in names]
399
400                if len(paths) > 1:
401                    logger.warning(f"More than 1 file encountered in the zipped export: {','.join([str(path) for path in paths])}, only returning first file")
402                logger.success(f"Enrichment results export with task ID: {task_id} has successfully finished.")
403
404                return paths[0]
405
406            waitable = WorkflowExecutionTaskWaitable[Path](
407                workflow_execution_id=workflow_execution_id, on_complete=on_complete, task_template_name=WorkflowTaskTemplateName.ENPI_APP_ENRICHMENT_EXPORT
408            )
409            return Execution(wait=waitable.wait_and_return_result, check_execution_state=waitable.check_execution_state)

Run an export of a Enrichment operation result and download the TSV file.

Arguments:

enrichment_run_id (enpi_api.l2.types.enrichment.EnrichmentRunId): Enrichment run ID.
operation (str): Name of the operation to export.
mode (enpi_api.l2.types.enrichment.EnrichmentExportMode): Mode in which export will be run.
tag_ids (list[enpi_api.l2.types.tag.TagId]): List of tags to be included in the export.
limit (int | None): If specified, the export will contain only the first N clusters, N being the value passed as this param.
output_directory (str | Path | None): The directory where to download the TSV file. Defaults to a unique temporary directory.

Returns:

enpi_api.l2.types.execution.Execution[Path]: An awaitable that returns the path to the downloaded TSV file containing the sequences.

Raises:

enpi_api.l2.types.api_error.ApiError: If API request fails.

Example:

with EnpiApiClient() as enpi_client:
    tsv_path = enpi_client.enrichment_api.export_as_tsv(
        enrichment_run_id=EnrichmentRunId(42),
        operation="Result",
        mode=EnrichmentExportMode.REPRESENTATIVES,
        tag_ids=[SequenceTags.Cdr3AminoAcids, SequenceTags.Chain],
        limit=50,
        output_directory="/results",
    ).wait()

def export_as_df( self, enrichment_run_id: enpi_api.l2.types.enrichment.EnrichmentRunId, operation: str, mode: enpi_api.l2.types.enrichment.EnrichmentExportMode, tag_ids: list[enpi_api.l2.types.tag.TagId], limit: int | None = None) -> enpi_api.l2.types.execution.Execution[DataFrame]: View Source

411    def export_as_df(
412        self,
413        enrichment_run_id: EnrichmentRunId,
414        operation: str,
415        mode: EnrichmentExportMode,
416        tag_ids: list[TagId],
417        limit: int | None = None,
418    ) -> Execution[pd.DataFrame]:
419        """Runs an Export of a Enrichment Operation results and loads it as a pandas DataFrame.
420
421        Args:
422            enrichment_run_id (enpi_api.l2.types.enrichment.EnrichmentRunId): Enrichment run ID.
423            operation (str): Name of the operation to export.
424            mode (enpi_api.l2.types.enrichment.EnrichmentExportMode): Mode in which export will be run.
425            tag_ids (list[enpi_api.l2.types.tag.TagId]): List of tags to be included in the export.
426            limit (int | None):
427                If specified, the export will contain only the first N clusters, N being the value passed as this param.
428
429        Returns:
430            enpi_api.l2.types.execution.Execution[Path]: An awaitable that returns the path to the downloaded TSV file containing the sequences.
431
432        Raises:
433            enpi_api.l2.types.api_error.ApiError: If API request fails.
434
435        Example:
436            ```python
437            with EnpiApiClient() as enpi_client:
438                tsv_path = enpi_client.enrichment_api.export_as_df(
439                    enrichment_run_id=EnrichmentRunId(42),
440                    operation="Result",
441                    mode=EnrichmentExportMode.REPRESENTATIVES,
442                    tag_ids=[SequenceTags.Cdr3AminoAcids, SequenceTags.Chain],
443                    limit=50
444                ).wait()
445            ```
446        """
447
448        export_tsv = self.export_as_tsv(enrichment_run_id, operation, mode, tag_ids, limit)
449
450        def wait() -> pd.DataFrame:
451            file_path = export_tsv.wait()
452            return pd.read_csv(file_path, delimiter="\t")
453
454        return Execution(wait=wait, check_execution_state=export_tsv.check_execution_state)

Runs an Export of a Enrichment Operation results and loads it as a pandas DataFrame.

Arguments:

enrichment_run_id (enpi_api.l2.types.enrichment.EnrichmentRunId): Enrichment run ID.
operation (str): Name of the operation to export.
mode (enpi_api.l2.types.enrichment.EnrichmentExportMode): Mode in which export will be run.
tag_ids (list[enpi_api.l2.types.tag.TagId]): List of tags to be included in the export.
limit (int | None): If specified, the export will contain only the first N clusters, N being the value passed as this param.

Returns:

enpi_api.l2.types.execution.Execution[Path]: An awaitable that returns the path to the downloaded TSV file containing the sequences.

Raises:

enpi_api.l2.types.api_error.ApiError: If API request fails.

Example:

with EnpiApiClient() as enpi_client:
    tsv_path = enpi_client.enrichment_api.export_as_df(
        enrichment_run_id=EnrichmentRunId(42),
        operation="Result",
        mode=EnrichmentExportMode.REPRESENTATIVES,
        tag_ids=[SequenceTags.Cdr3AminoAcids, SequenceTags.Chain],
        limit=50
    ).wait()