Module `phc.easy.projects`

Expand source code

import inspect
from typing import Optional

import pandas as pd
from funcy import memoize
from phc.easy.abstract.paging_api_item import PagingApiItem, PagingApiOptions
from phc.easy.auth import Auth
from phc.easy.util import without_keys
from phc.errors import ApiError
from pmap import pmap

SEARCH_COLUMNS = ["name", "description", "id"]


def join_strings(row):
    return " ".join([value for value in row if type(value) == str]).lower()


class ProjectListOptions(PagingApiOptions):
    name: Optional[str]


class Project(PagingApiItem):
    @staticmethod
    def resource_path():
        return "projects"

    @staticmethod
    def params_class():
        return ProjectListOptions

    @classmethod
    @memoize
    def get_data_frame(
        cls,
        name: Optional[str] = None,
        auth_args: Auth = Auth.shared(),
        max_pages: Optional[int] = None,
        page_size: Optional[int] = None,
        log: bool = False,
        show_progress: bool = False,
        account: Optional[str] = None,
    ):
        """Execute a request for projects

        ## Parameters

        Query: `phc.easy.projects.ProjectListOptions`

        Execution: `phc.easy.query.Query.execute_paging_api`
        """

        if page_size is None:
            # Projects do not have much data so use a higher page size
            page_size = 100

        get_data_frame = super().get_data_frame

        auth = Auth(auth_args)

        get_data_frame_args = without_keys(
            cls._get_current_args(inspect.currentframe(), locals()),
            ["auth_args", "account", "show_progress"],
        )

        def get_projects_for_account(account: dict):
            try:
                df = get_data_frame(
                    ignore_cache=True,
                    all_results=max_pages is None,
                    auth_args=auth.customized({"account": account["id"]}),
                    show_progress=show_progress,
                    **get_data_frame_args,
                )
                df["account"] = account["id"]

                return df

            except ApiError as e:
                message = e.response.get("error", "Unknown API error")
                print(f"Skipping \"{account['id']}\" due to \"{message}\"")

                return pd.DataFrame()

        if account:
            return get_projects_for_account({"id": account})

        return pd.concat(
            list(pmap(get_projects_for_account, auth.accounts()))
        ).reset_index(drop=True)

    @staticmethod
    def find(
        search: str,
        account: Optional[str] = None,
        auth_args: Auth = Auth.shared(),
    ):
        """Search for a project using given criteria and return results as a data frame

        Attributes
        ----------
        search : str
            Part of a project's id, name, or description to search for

        auth_args : Any
            The authenication to use for the account and project (defaults to shared)
        """
        projects = Project.get_data_frame(auth_args=auth_args, account=account)
        text = projects[SEARCH_COLUMNS].agg(join_strings, axis=1)
        return projects[text.str.contains(search.lower())]

    @staticmethod
    def set_current(
        search: str, account: Optional[str] = None, auth: Auth = Auth.shared()
    ):
        """Search for a project using given criteria, set it to the authentication
        object, and return the matching projects as a data frame

        Attributes
        ----------
        search : str
            Part of a project's id, name, or description to search for

        auth : Auth
            The authenication to update for the account and project (defaults to shared)
        """
        matches = Project.find(search, account=account, auth_args=auth)

        if len(matches) > 1:
            print("Multiple projects found. Try a more specific search")
        elif len(matches) == 0:
            print(f'No matches found for search "{search}"')
        else:
            project = matches.iloc[0]
            # Uses private method since this is a special case
            auth.update({"account": project.account, "project_id": project.id})

        return matches

Functions

def join_strings(row)

Expand source code

def join_strings(row):
    return " ".join([value for value in row if type(value) == str]).lower()

Classes

class Project

Expand source code

class Project(PagingApiItem):
    @staticmethod
    def resource_path():
        return "projects"

    @staticmethod
    def params_class():
        return ProjectListOptions

    @classmethod
    @memoize
    def get_data_frame(
        cls,
        name: Optional[str] = None,
        auth_args: Auth = Auth.shared(),
        max_pages: Optional[int] = None,
        page_size: Optional[int] = None,
        log: bool = False,
        show_progress: bool = False,
        account: Optional[str] = None,
    ):
        """Execute a request for projects

        ## Parameters

        Query: `phc.easy.projects.ProjectListOptions`

        Execution: `phc.easy.query.Query.execute_paging_api`
        """

        if page_size is None:
            # Projects do not have much data so use a higher page size
            page_size = 100

        get_data_frame = super().get_data_frame

        auth = Auth(auth_args)

        get_data_frame_args = without_keys(
            cls._get_current_args(inspect.currentframe(), locals()),
            ["auth_args", "account", "show_progress"],
        )

        def get_projects_for_account(account: dict):
            try:
                df = get_data_frame(
                    ignore_cache=True,
                    all_results=max_pages is None,
                    auth_args=auth.customized({"account": account["id"]}),
                    show_progress=show_progress,
                    **get_data_frame_args,
                )
                df["account"] = account["id"]

                return df

            except ApiError as e:
                message = e.response.get("error", "Unknown API error")
                print(f"Skipping \"{account['id']}\" due to \"{message}\"")

                return pd.DataFrame()

        if account:
            return get_projects_for_account({"id": account})

        return pd.concat(
            list(pmap(get_projects_for_account, auth.accounts()))
        ).reset_index(drop=True)

    @staticmethod
    def find(
        search: str,
        account: Optional[str] = None,
        auth_args: Auth = Auth.shared(),
    ):
        """Search for a project using given criteria and return results as a data frame

        Attributes
        ----------
        search : str
            Part of a project's id, name, or description to search for

        auth_args : Any
            The authenication to use for the account and project (defaults to shared)
        """
        projects = Project.get_data_frame(auth_args=auth_args, account=account)
        text = projects[SEARCH_COLUMNS].agg(join_strings, axis=1)
        return projects[text.str.contains(search.lower())]

    @staticmethod
    def set_current(
        search: str, account: Optional[str] = None, auth: Auth = Auth.shared()
    ):
        """Search for a project using given criteria, set it to the authentication
        object, and return the matching projects as a data frame

        Attributes
        ----------
        search : str
            Part of a project's id, name, or description to search for

        auth : Auth
            The authenication to update for the account and project (defaults to shared)
        """
        matches = Project.find(search, account=account, auth_args=auth)

        if len(matches) > 1:
            print("Multiple projects found. Try a more specific search")
        elif len(matches) == 0:
            print(f'No matches found for search "{search}"')
        else:
            project = matches.iloc[0]
            # Uses private method since this is a special case
            auth.update({"account": project.account, "project_id": project.id})

        return matches

Ancestors

PagingApiItem

Static methods

def find(search: str, account: Optional[str] = None, auth_args: Auth = <phc.easy.auth.Auth object>)

Search for a project using given criteria and return results as a data frame

Attributes

search : str: Part of a project's id, name, or description to search for
auth_args : Any: The authenication to use for the account and project (defaults to shared)

Expand source code

@staticmethod
def find(
    search: str,
    account: Optional[str] = None,
    auth_args: Auth = Auth.shared(),
):
    """Search for a project using given criteria and return results as a data frame

    Attributes
    ----------
    search : str
        Part of a project's id, name, or description to search for

    auth_args : Any
        The authenication to use for the account and project (defaults to shared)
    """
    projects = Project.get_data_frame(auth_args=auth_args, account=account)
    text = projects[SEARCH_COLUMNS].agg(join_strings, axis=1)
    return projects[text.str.contains(search.lower())]

def get_data_frame(cls, name: Optional[str] = None, auth_args: Auth = <phc.easy.auth.Auth object>, max_pages: Optional[int] = None, page_size: Optional[int] = None, log: bool = False, show_progress: bool = False, account: Optional[str] = None)

Execute a request for projects

Parameters

Query: ProjectListOptions

Execution: Query.execute_paging_api()

Expand source code

@classmethod
@memoize
def get_data_frame(
    cls,
    name: Optional[str] = None,
    auth_args: Auth = Auth.shared(),
    max_pages: Optional[int] = None,
    page_size: Optional[int] = None,
    log: bool = False,
    show_progress: bool = False,
    account: Optional[str] = None,
):
    """Execute a request for projects

    ## Parameters

    Query: `phc.easy.projects.ProjectListOptions`

    Execution: `phc.easy.query.Query.execute_paging_api`
    """

    if page_size is None:
        # Projects do not have much data so use a higher page size
        page_size = 100

    get_data_frame = super().get_data_frame

    auth = Auth(auth_args)

    get_data_frame_args = without_keys(
        cls._get_current_args(inspect.currentframe(), locals()),
        ["auth_args", "account", "show_progress"],
    )

    def get_projects_for_account(account: dict):
        try:
            df = get_data_frame(
                ignore_cache=True,
                all_results=max_pages is None,
                auth_args=auth.customized({"account": account["id"]}),
                show_progress=show_progress,
                **get_data_frame_args,
            )
            df["account"] = account["id"]

            return df

        except ApiError as e:
            message = e.response.get("error", "Unknown API error")
            print(f"Skipping \"{account['id']}\" due to \"{message}\"")

            return pd.DataFrame()

    if account:
        return get_projects_for_account({"id": account})

    return pd.concat(
        list(pmap(get_projects_for_account, auth.accounts()))
    ).reset_index(drop=True)

def params_class()

Inherited from: PagingApiItem.params_class

Returns a pydantic type that validates and transforms the params with dict()

Expand source code

@staticmethod
def params_class():
    return ProjectListOptions

def process_params(params: dict) ‑> dict

Inherited from: PagingApiItem.process_params

Validates and transforms the API query parameters

def resource_path()

Inherited from: PagingApiItem.resource_path

Returns the API url name for retrieval

Expand source code

@staticmethod
def resource_path():
    return "projects"

def set_current(search: str, account: Optional[str] = None, auth: Auth = <phc.easy.auth.Auth object>)

Search for a project using given criteria, set it to the authentication object, and return the matching projects as a data frame

Attributes

search : str: Part of a project's id, name, or description to search for
auth : Auth: The authenication to update for the account and project (defaults to shared)

Expand source code

@staticmethod
def set_current(
    search: str, account: Optional[str] = None, auth: Auth = Auth.shared()
):
    """Search for a project using given criteria, set it to the authentication
    object, and return the matching projects as a data frame

    Attributes
    ----------
    search : str
        Part of a project's id, name, or description to search for

    auth : Auth
        The authenication to update for the account and project (defaults to shared)
    """
    matches = Project.find(search, account=account, auth_args=auth)

    if len(matches) > 1:
        print("Multiple projects found. Try a more specific search")
    elif len(matches) == 0:
        print(f'No matches found for search "{search}"')
    else:
        project = matches.iloc[0]
        # Uses private method since this is a special case
        auth.update({"account": project.account, "project_id": project.id})

    return matches

def transform_results(data_frame: pandas.core.frame.DataFrame, **expand_args)

Inherited from: PagingApiItem.transform_results

Transform data frame batch

class ProjectListOptions (**data: Any)

Usage docs: https://docs.pydantic.dev/2.10/concepts/models/

A base class for creating Pydantic models.

Attributes

__class_vars__: The names of the class variables defined on the model.
__private_attributes__: Metadata about the private attributes of the model.
__signature__: The synthesized __init__ [Signature][inspect.Signature] of the model.
__pydantic_complete__: Whether model building is completed, or if there are still undefined fields.
__pydantic_core_schema__: The core schema of the model.
__pydantic_custom_init__: Whether the model has a custom __init__ function.
__pydantic_decorators__: Metadata containing the decorators defined on the model. This replaces Model.__validators__ and Model.__root_validators__ from Pydantic V1.
__pydantic_generic_metadata__: Metadata for generic models; contains data used for a similar purpose to args, origin, parameters in typing-module generics. May eventually be replaced by these.
__pydantic_parent_namespace__: Parent namespace of the model, used for automatic rebuilding of models.
__pydantic_post_init__: The name of the post-init method for the model, if defined.
__pydantic_root_model__: Whether the model is a [RootModel][pydantic.root_model.RootModel].
__pydantic_serializer__: The pydantic-core SchemaSerializer used to dump instances of the model.
__pydantic_validator__: The pydantic-core SchemaValidator used to validate instances of the model.
__pydantic_fields__: A dictionary of field names and their corresponding [FieldInfo][pydantic.fields.FieldInfo] objects.
__pydantic_computed_fields__: A dictionary of computed field names and their corresponding [ComputedFieldInfo][pydantic.fields.ComputedFieldInfo] objects.
__pydantic_extra__: A dictionary containing extra values, if [extra][pydantic.config.ConfigDict.extra] is set to 'allow'.
__pydantic_fields_set__: The names of fields explicitly set during instantiation.
__pydantic_private__: Values of private attributes set on the model instance.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

Expand source code

class ProjectListOptions(PagingApiOptions):
    name: Optional[str]

Ancestors

PagingApiOptions
pydantic.main.BaseModel

Class variables

var model_config
var name : Optional[str]

Methods

def model_dump(self): Inherited from: PagingApiOptions.model_dump

Usage docs: https://docs.pydantic.dev/2.10/concepts/serialization/#modelmodel_dump …