Module phc.easy.abstract.paging_api_item

Expand source code 
from typing import Any, Union, List
import inspect
from enum import Enum

import pandas as pd
from phc.easy.frame import Frame
from phc.easy.query import Query
from pydantic import BaseModel
from toolz import groupby

EXECUTE_QUERY_ARGS = inspect.getfullargspec(Query.execute_paging_api).args
EXPAND_ARGS = inspect.getfullargspec(Frame.expand).args


class PagingApiOptions(BaseModel):
    @staticmethod
    def transform(key, value):
        return (key, value)

    def model_dump(self):
        raw = super().model_dump()

        def preprocess_value(v):
            if isinstance(v, Enum):
                return v.value

            return v

        return dict(
            [
                self.__class__.transform(k, preprocess_value(v))
                for k, v in raw.items()
            ]
        )


def split_kw_args(
    args: dict,
    blacklist: List[str] = [],
    additional_expand_keys: List[str] = [],
):
    def value(pair):
        if pair[0] in EXECUTE_QUERY_ARGS:
            return "execute"
        elif pair[0] in [*EXPAND_ARGS, *additional_expand_keys]:
            return "expand"
        else:
            return "query"

    for blacklist_key in blacklist:
        if blacklist_key in args.keys():
            raise ValueError(f"'{blacklist_key}' not supported.")

    return {k: dict(v) for k, v in groupby(value, args.items()).items()}


class PagingApiItem:
    @staticmethod
    def resource_path() -> str:
        "Returns the API url name for retrieval"
        return ""

    @staticmethod
    def params_class() -> type:
        "Returns a pydantic type that validates and transforms the params with dict()"
        return PagingApiOptions

    @classmethod
    def process_params(cls, params: dict) -> dict:
        "Validates and transforms the API query parameters"
        return cls.params_class()(**params).model_dump()

    @staticmethod
    def transform_results(data_frame: pd.DataFrame, **expand_args):
        "Transform data frame batch"
        return data_frame

    @staticmethod
    def response_to_items(data: Union[list, dict]) -> list:
        return data.get("items", [])

    @staticmethod
    def execute_args() -> dict:
        return {}

    @staticmethod
    def _get_current_args(frame: Any, local_vars: dict):
        """Helper function for getting all arguments to the current function as a dictionary"""
        EXCEPTIONS = ["cls", "frame"]
        all_arg_names = inspect.getargvalues(frame).args
        return {k: local_vars[k] for k in all_arg_names if k not in EXCEPTIONS}

    @classmethod
    def get_data_frame(cls, **kw_args):
        expand_keys = [
            k
            for k in inspect.getfullargspec(cls.transform_results).args
            if k not in ["frame", "expand_keys"]
        ]
        overrides = cls.execute_args()
        split_args = split_kw_args(
            {**kw_args, **overrides},
            blacklist=["item_key"],
            additional_expand_keys=expand_keys,
        )
        params, expand_args, execute_options = (
            cls.process_params(split_args.get("query", {})),
            split_args.get("expand", {}),
            split_args.get("execute", {}),
        )

        def transform(df: pd.DataFrame):
            if len(df) == 0:
                return df

            return cls.transform_results(df, params=params, **expand_args)

        df = Query.execute_paging_api(
            cls.resource_path(),
            params,
            **execute_options,
            transform=transform,
            response_to_items=cls.response_to_items,
        )

        return df

Functions

def split_kw_args(args: dict, blacklist: List[str] = [], additional_expand_keys: List[str] = [])
Expand source code 
def split_kw_args(
    args: dict,
    blacklist: List[str] = [],
    additional_expand_keys: List[str] = [],
):
    def value(pair):
        if pair[0] in EXECUTE_QUERY_ARGS:
            return "execute"
        elif pair[0] in [*EXPAND_ARGS, *additional_expand_keys]:
            return "expand"
        else:
            return "query"

    for blacklist_key in blacklist:
        if blacklist_key in args.keys():
            raise ValueError(f"'{blacklist_key}' not supported.")

    return {k: dict(v) for k, v in groupby(value, args.items()).items()}

Classes

class PagingApiItem
Expand source code 
class PagingApiItem:
    @staticmethod
    def resource_path() -> str:
        "Returns the API url name for retrieval"
        return ""

    @staticmethod
    def params_class() -> type:
        "Returns a pydantic type that validates and transforms the params with dict()"
        return PagingApiOptions

    @classmethod
    def process_params(cls, params: dict) -> dict:
        "Validates and transforms the API query parameters"
        return cls.params_class()(**params).model_dump()

    @staticmethod
    def transform_results(data_frame: pd.DataFrame, **expand_args):
        "Transform data frame batch"
        return data_frame

    @staticmethod
    def response_to_items(data: Union[list, dict]) -> list:
        return data.get("items", [])

    @staticmethod
    def execute_args() -> dict:
        return {}

    @staticmethod
    def _get_current_args(frame: Any, local_vars: dict):
        """Helper function for getting all arguments to the current function as a dictionary"""
        EXCEPTIONS = ["cls", "frame"]
        all_arg_names = inspect.getargvalues(frame).args
        return {k: local_vars[k] for k in all_arg_names if k not in EXCEPTIONS}

    @classmethod
    def get_data_frame(cls, **kw_args):
        expand_keys = [
            k
            for k in inspect.getfullargspec(cls.transform_results).args
            if k not in ["frame", "expand_keys"]
        ]
        overrides = cls.execute_args()
        split_args = split_kw_args(
            {**kw_args, **overrides},
            blacklist=["item_key"],
            additional_expand_keys=expand_keys,
        )
        params, expand_args, execute_options = (
            cls.process_params(split_args.get("query", {})),
            split_args.get("expand", {}),
            split_args.get("execute", {}),
        )

        def transform(df: pd.DataFrame):
            if len(df) == 0:
                return df

            return cls.transform_results(df, params=params, **expand_args)

        df = Query.execute_paging_api(
            cls.resource_path(),
            params,
            **execute_options,
            transform=transform,
            response_to_items=cls.response_to_items,
        )

        return df

Subclasses

Static methods

def execute_args() ‑> dict
Expand source code 
@staticmethod
def execute_args() -> dict:
    return {}
def get_data_frame(**kw_args)
Expand source code 
@classmethod
def get_data_frame(cls, **kw_args):
    expand_keys = [
        k
        for k in inspect.getfullargspec(cls.transform_results).args
        if k not in ["frame", "expand_keys"]
    ]
    overrides = cls.execute_args()
    split_args = split_kw_args(
        {**kw_args, **overrides},
        blacklist=["item_key"],
        additional_expand_keys=expand_keys,
    )
    params, expand_args, execute_options = (
        cls.process_params(split_args.get("query", {})),
        split_args.get("expand", {}),
        split_args.get("execute", {}),
    )

    def transform(df: pd.DataFrame):
        if len(df) == 0:
            return df

        return cls.transform_results(df, params=params, **expand_args)

    df = Query.execute_paging_api(
        cls.resource_path(),
        params,
        **execute_options,
        transform=transform,
        response_to_items=cls.response_to_items,
    )

    return df
def params_class() ‑> type

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

Expand source code 
@staticmethod
def params_class() -> type:
    "Returns a pydantic type that validates and transforms the params with dict()"
    return PagingApiOptions
def process_params(params: dict) ‑> dict

Validates and transforms the API query parameters

Expand source code 
@classmethod
def process_params(cls, params: dict) -> dict:
    "Validates and transforms the API query parameters"
    return cls.params_class()(**params).model_dump()
def resource_path() ‑> str

Returns the API url name for retrieval

Expand source code 
@staticmethod
def resource_path() -> str:
    "Returns the API url name for retrieval"
    return ""
def response_to_items(data: Union[list, dict]) ‑> list
Expand source code 
@staticmethod
def response_to_items(data: Union[list, dict]) -> list:
    return data.get("items", [])
def transform_results(data_frame: pandas.core.frame.DataFrame, **expand_args)

Transform data frame batch

Expand source code 
@staticmethod
def transform_results(data_frame: pd.DataFrame, **expand_args):
    "Transform data frame batch"
    return data_frame
class PagingApiOptions (**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 PagingApiOptions(BaseModel):
    @staticmethod
    def transform(key, value):
        return (key, value)

    def model_dump(self):
        raw = super().model_dump()

        def preprocess_value(v):
            if isinstance(v, Enum):
                return v.value

            return v

        return dict(
            [
                self.__class__.transform(k, preprocess_value(v))
                for k, v in raw.items()
            ]
        )

Ancestors

  • pydantic.main.BaseModel

Subclasses

Class variables

var model_config

Static methods

def transform(key, value)
Expand source code 
@staticmethod
def transform(key, value):
    return (key, value)

Methods

def model_dump(self)

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

Generate a dictionary representation of the model, optionally specifying which fields to include or exclude.

Args

mode
The mode in which to_python should run. If mode is 'json', the output will only contain JSON serializable types. If mode is 'python', the output may contain non-JSON-serializable Python objects.
include
A set of fields to include in the output.
exclude
A set of fields to exclude from the output.
context
Additional context to pass to the serializer.
by_alias
Whether to use the field's alias in the dictionary key if defined.
exclude_unset
Whether to exclude fields that have not been explicitly set.
exclude_defaults
Whether to exclude fields that are set to their default value.
exclude_none
Whether to exclude fields that have a value of None.
round_trip
If True, dumped values should be valid as input for non-idempotent types such as Json[T].
warnings
How to handle serialization errors. False/"none" ignores them, True/"warn" logs errors, "error" raises a [PydanticSerializationError][pydantic_core.PydanticSerializationError].
serialize_as_any
Whether to serialize fields with duck-typing serialization behavior.

Returns

A dictionary representation of the model.

Expand source code 
def model_dump(self):
    raw = super().model_dump()

    def preprocess_value(v):
        if isinstance(v, Enum):
            return v.value

        return v

    return dict(
        [
            self.__class__.transform(k, preprocess_value(v))
            for k, v in raw.items()
        ]
    )