project

Define a project.

Classes:

Name	Description
Project	A project is used to organize your artifacts.

Functions:

Name	Description
set_project	Set the current project.
get_project	Get the current project.

Project(model: ProjectSelectOne | ProjectSelect)

A project is used to organize your artifacts.

Methods:

Name	Description
create_or_get	Create project with the given name and description. Load project if the given name already exists.
load_all	Load all projects.
update	Update the current project.
__enter__	Set Project.CURRENT to this project.
__exit__	Restore previous Project.CURRENT.
list_fitted_datasets	List all datasets in this project.
list_trained_models	List all trained models in this project.
list_explanations	List all explanations in this project.
list_computed_statistics	List all statistics in this project.
delete	Delete this project.

Attributes:

Name	Type	Description
CURRENT	ContextVar[Project]
model

Source code in src/xpdeep/project.py

def __init__(self, model: ProjectSelectOne | ProjectSelect) -> None:
    self.model = model

CURRENT: ContextVar[Project] = ContextVar('CURRENT_PROJECT')

model = model

create_or_get(*, name: str, description: str = '') -> Project

Create project with the given name and description. Load project if the given name already exists.

Parameters:

Name	Type	Description	Default
name	str	Name of the project to create. If a project within this name already exists, it will be loaded	required
description	str	Description of the new created project. Ignored in case of project loading.	""

Source code in src/xpdeep/project.py

@staticmethod
@initialized_client_verification
def create_or_get(*, name: str, description: str = "") -> Project:
    """Create project with the given name and description. Load project if the given name already exists.

    Parameters
    ----------
    name : str
        Name of the project to create. If a project within this name already exists, it will be loaded
    description : str, default ""
        Description of the new created project. Ignored in case of project loading.
    """
    try:
        project_model = next(project for project in Project.load_all() if project.model.name == name).model
    except StopIteration:
        try:
            project_model = handle_api_validation_errors(
                insert_one_project.sync(
                    body=ProjectInsert(
                        name=name,
                        description=description,
                    ),
                    client=ClientFactory.CURRENT.get()(),
                ),
            )
        except UnexpectedStatus as err:
            message = (
                f"A Project named : `{name}` already exists on the instance you are using. "
                f"As you have no access rights to this project, you cannot neither get it or create a new "
                f"one with same name."
            )
            raise ApiError(message=message) from err

    return Project(project_model)

load_all() -> list[Project]

Load all projects.

Source code in src/xpdeep/project.py

@classmethod
@initialized_client_verification
def load_all(cls) -> list[Project]:
    """Load all projects."""
    with ClientFactory.CURRENT.get()() as client:
        all_project_models = _get_all_objects_from_paginated_route(select_projects.sync, client=client)
    return [cls(project_model) for project_model in all_project_models]

update(*, name: str | None = None, description: str | None = None) -> None

Update the current project.

Parameters:

Name	Type	Description	Default
name	str | None	New project's name.	None
description	str | None	New project's description.	None

Source code in src/xpdeep/project.py

@initialized_client_verification
def update(self, *, name: str | None = None, description: str | None = None) -> None:
    """Update the current project.

    Parameters
    ----------
    name : str | None, default None
        New project's name.
    description : str | None, default None
        New project's description.
    """
    if name is not None or description is not None:
        self.model = handle_api_validation_errors(
            update_project.sync(
                self.model.id,
                client=ClientFactory.CURRENT.get()(),
                body=ProjectUpdate(name=name, description=description),
            ),
        )

    if name is not None:
        self.model.name = name

    if description is not None:
        self.model.description = description

__enter__() -> Self

Set Project.CURRENT to this project.

Source code in src/xpdeep/project.py

def __enter__(self) -> Self:
    """Set ``Project.CURRENT`` to this project."""
    self.token = Project.CURRENT.set(self)

    return self

__exit__(exc_type: type[BaseException] | None, exc_val: BaseException | None, exc_tb: TracebackType | None) -> None

Restore previous Project.CURRENT.

Source code in src/xpdeep/project.py

def __exit__(
    self, exc_type: type[BaseException] | None, exc_val: BaseException | None, exc_tb: TracebackType | None
) -> None:
    """Restore previous Project.CURRENT."""
    Project.CURRENT.reset(self.token)

list_fitted_datasets() -> list[FittedParquetDataset]

List all datasets in this project.

Source code in src/xpdeep/project.py

@initialized_client_verification
def list_fitted_datasets(self) -> list[FittedParquetDataset]:
    """List all datasets in this project."""
    from xpdeep.dataset.parquet_dataset import FittedParquetDataset  # noqa: PLC0415

    with ClientFactory.CURRENT.get()() as client:
        parquet_dataset_selects = _get_all_objects_from_paginated_route(
            select_parquet_dataset_artifacts.sync, self.model.id, client=client
        )
        with ThreadPoolExecutor(max_workers=8) as executor:
            parquet_dataset_artifact_select_ones = [
                handle_api_validation_errors(future.result())
                for future in as_completed(
                    executor.submit(
                        select_one_parquet_dataset_artifact.sync,
                        self.model.id,
                        parquet_dataset_select.id,
                        client=client,
                    )
                    for parquet_dataset_select in parquet_dataset_selects
                )
            ]

        fitted_parquet_datasets = []

        for parquet_dataset_artifact_select_one in parquet_dataset_artifact_select_ones:
            fitted_parquet_dataset_ = FittedParquetDataset(
                name=parquet_dataset_artifact_select_one.name,
                path=parquet_dataset_artifact_select_one.path,
                storage_options=parquet_dataset_artifact_select_one.storage_options.to_dict(),
                fitted_schema=FittedSchema.from_model(parquet_dataset_artifact_select_one.schema.to_dict()),
            )

            fitted_parquet_dataset_._artifact_id = parquet_dataset_artifact_select_one.id  # noqa : SLF001
            fitted_parquet_datasets.append(fitted_parquet_dataset_)

        return fitted_parquet_datasets

list_trained_models() -> list[TrainedModelArtifact]

List all trained models in this project.

Source code in src/xpdeep/project.py

@initialized_client_verification
def list_trained_models(self) -> list[TrainedModelArtifact]:
    """List all trained models in this project."""
    from xpdeep.model.trained_model_artifact import TrainedModelArtifact  # noqa: PLC0415

    with ClientFactory.CURRENT.get()() as client:
        trained_model_selects = _get_all_objects_from_paginated_route(
            select_trained_model_artifacts.sync, self.model.id, client=client
        )
        with ThreadPoolExecutor(max_workers=8) as executor:
            trained_model_artifact_select_ones = [
                handle_api_validation_errors(future.result())
                for future in as_completed(
                    executor.submit(
                        select_one_trained_model_artifact.sync,
                        self.model.id,
                        trained_model_select.id,
                        client=client,
                    )
                    for trained_model_select in trained_model_selects
                )
            ]
            return [
                TrainedModelArtifact.from_dict(trained_model_select_one.to_dict())
                for trained_model_select_one in trained_model_artifact_select_ones
            ]

list_explanations() -> list[ExplanationArtifact]

List all explanations in this project.

Source code in src/xpdeep/project.py

@initialized_client_verification
def list_explanations(self) -> list[ExplanationArtifact]:
    """List all explanations in this project."""
    from xpdeep.explain.explanation_artifact import ExplanationArtifact  # noqa: PLC0415

    with ClientFactory.CURRENT.get()() as client:
        explanation_artifact_selects = _get_all_objects_from_paginated_route(
            select_explanation_artifacts.sync, self.model.id, client=client
        )
        with ThreadPoolExecutor(max_workers=8) as executor:
            explanation_artifact_select_ones = [
                handle_api_validation_errors(future.result())
                for future in as_completed(
                    executor.submit(
                        select_one_explanation_artifact.sync,
                        self.model.id,
                        explanation_artifact_select.id,
                        client=client,
                    )
                    for explanation_artifact_select in explanation_artifact_selects
                )
            ]
            return [
                ExplanationArtifact.from_dict(explanation_artifact_select_one.to_dict())
                for explanation_artifact_select_one in explanation_artifact_select_ones
            ]

list_computed_statistics(dataset: FittedParquetDataset) -> list[ExplanationStatisticSelect]

List all statistics in this project.

Parameters:

Name	Type	Description	Default
dataset	FittedParquetDataset	Dataset on which the statistics were computed.	required

Source code in src/xpdeep/project.py

@initialized_client_verification
def list_computed_statistics(self, dataset: FittedParquetDataset) -> list[ExplanationStatisticSelect]:
    """List all statistics in this project.

    Parameters
    ----------
    dataset : FittedParquetDataset
        Dataset on which the statistics were computed.
    """
    with ClientFactory.CURRENT.get()() as client:
        return _get_all_objects_from_paginated_route(
            select_explanation_statistics.sync,
            self.model.id,
            client=client,
            dataset_artifact_id=dataset.artifact_id(),
        )

delete() -> None

Delete this project.

Also deletes all artifacts in this project recursively. WARNING: not recoverable

Source code in src/xpdeep/project.py

@initialized_client_verification
def delete(self) -> None:
    """Delete this project.

    Also deletes all artifacts in this project recursively. WARNING: not recoverable
    """
    client_factory = ClientFactory.CURRENT.get()
    with client_factory() as client:
        delete_project.sync(self.model.id, client=client)
    del self.model

set_project(project: Project) -> None

Set the current project.

Source code in src/xpdeep/project.py

def set_project(project: Project) -> None:
    """Set the current project."""
    Project.CURRENT.set(project)

get_project() -> Project | None

Get the current project.

Source code in src/xpdeep/project.py

def get_project() -> Project | None:
    """Get the current project."""
    try:
        return Project.CURRENT.get()
    except LookupError:
        return None