xpdeep

Xpdeep.

Modules:

Name	Description
client_factory	The client factory module is responsible for creating a client instance.
dataset	Manage data upload to be processed by xpdeep.
explain	Define the explain package, used to compute explanations.
filtering	Filtering module.
initialize	Initialize the API.
metric	Metrics to use for training or explanations.
model	The model package provides tools for working with Xpdeep models.
project	Define a project.
trainer	Define the trainer package, used to train the XpdeepModel.
utils	Utils package.

Classes:

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

Functions:

Name	Description
init	Initialize the Xpdeep API with an API key.
get_project	Get the current project.
set_project	Set the current project.

__all__ = ('Project', 'get_project', 'init', 'set_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

init(api_key: str, api_url: str = 'https://xpdeep.com/api') -> None

Initialize the Xpdeep API with an API key.

To generate an API key, please refer to the documentation.

https://docs.xpdeep.com/latest/Getting%20started/installation/#get-an-api-key

Source code in src/xpdeep/initialize.py

def init(
    api_key: str,
    api_url: str = "https://xpdeep.com/api",
) -> None:
    """Initialize the Xpdeep API with an API key.

    To generate an API key, please refer to the documentation.

    https://docs.xpdeep.com/latest/Getting%20started/installation/#get-an-api-key
    """
    ClientFactory.CURRENT.set(ClientFactory(api_key, api_url))
    with ClientFactory.CURRENT.get()() as client:
        try:
            response = handle_api_validation_errors(check_auth.sync(client=client))
        except UnexpectedStatus as exc:
            if exc.status_code == 401:  # noqa: PLR2004
                msg = (
                    "Please check your API key:\n"
                    "https://docs.xpdeep.com/latest/Getting%20started/installation/#get-an-api-key"
                )
                raise ApiError(msg) from exc
            if exc.status_code == 404:  # noqa: PLR2004
                msg = "Please check your API URL. Does it end with '/api'?"
                raise ApiError(msg) from exc
            raise
        if response is not None:
            logger.debug("authenticated: user_id=%s, tenant_id=%s", response.id, response.tenant_id)

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

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)