athena.client

Athena

Main api for executed modules

Attributes:

Name	Type	Description
fixture	Fixture	provider for test fixtures
infix	Fixture	wrapper for fixture that will inject Athena instance as the first argument
cache	Cache	persistent key (str) - value (str, int, float, bool) cache
context	Context	information about the runtime environment of the module
fake	Fake	generate randomized data
variable	ResourceFacade	get a variable by name
secret	ResourceFacade	get a secret by name

Source code in athena/client.py

class Athena:
    """Main api for executed modules

    Attributes:
        fixture (Fixture): provider for test fixtures
        infix (Fixture): wrapper for `fixture` that will inject `Athena` instance as the first argument
        cache (Cache): persistent key (`str`) - value (`str`, `int`, `float`, `bool`) cache
        context (Context): information about the runtime environment of the module
        fake (Fake): generate randomized data
        variable (ResourceFacade): get a variable by name
        secret (ResourceFacade): get a secret by name
    """
    def __init__(self,
        context: Context,
        session: AthenaSession,
        cache_values: dict
    ):
        self.__history: list[AthenaTrace | str] = []
        self.__pending_requests = {}
        self.__history_lookup_cache = {}
        self.__session = session
        self.fixture: Fixture = _Fixture()
        self.infix: Fixture = _InjectFixture(self.fixture, self)
        self.cache = Cache(cache_values)
        self.context = context
        self.fake = Fake()
        self.variable = ResourceFacade(
            lambda: self.__session.resource_loader.load_variables(self.context.root_path, self.context.module_path),
            'variable', self.context.environment)
        self.secret = ResourceFacade(
            lambda: self.__session.resource_loader.load_secrets(self.context.root_path, self.context.module_path),
            'secret', self.context.environment)

    def __client_pre_hook(self, trace_id: str) -> None:
        self.__history.append(trace_id)
        self.__pending_requests[trace_id] = len(self.__history) - 1

    def __client_post_hook(self, trace: AthenaTrace) -> None:
        if trace.id in self.__pending_requests:
            index = self.__pending_requests.pop(trace.id)
            if index < len(self.__history):
                self.__history[index] = trace
                return
        self.__history.append(trace)

    def client(self, base_build_request: Callable[[RequestBuilder], RequestBuilder] | None=None, name: str | None=None) -> Client:
        """Create a new client.

        Args:
            base_build_request (Callable[[RequestBuilder], [RequestBuilder]] | None): Function to configure all requests sent by the client.
            name (str | None): Optional name for client.

        Returns:
            Client: The configured client.
        """
        return Client(self.__session.session, self.__session.async_session, base_build_request, name, self.__client_pre_hook, self.__client_post_hook)

    def traces(self) -> list[AthenaTrace]:
        """Get all `AthenaTrace`s from the lifetime of this instance. 

        Returns:
            list[AthenaTrace]: List of traces.
        """
        return [i for i in self.__history if isinstance(i, AthenaTrace)]

    def trace(self, subject: AthenaTrace | RequestTrace | ResponseTrace | None=None) -> AthenaTrace:
        """Get the full `AthenaTrace` for a given request or response trace.

        Args:
            subject (AthenaTrace | RequestTrace | ResponseTrace | None): The trace to lookup. If no trace is provided, will return the most recent trace.

        Returns:
            AthenaTrace: The full `AthenaTrace` for the subject.

        Example:

            def run(athena: Athena):
                response = athena.client().get('https://example.com')
                trace = athena.trace(response)
                print(f'request completed in {trace.elapsed} seconds')
        """
        traces = self.traces()
        if subject is None:
            if len(traces) == 0:
                raise AthenaException(f"no completed traces in history")
            subject = traces[-1]

        is_trace = isinstance(subject, AthenaTrace)
        is_request_trace = isinstance(subject, RequestTrace)
        is_response_trace = isinstance(subject, ResponseTrace)
        if not (is_trace or is_request_trace or is_response_trace):
            raise AthenaException(f"unable to resolve parent for trace of type {type(subject).__name__}")

        if subject in self.__history_lookup_cache:
            return self.__history_lookup_cache[subject]
        trace = None
        for historic_trace in traces:
            if ((is_trace and historic_trace == subject)
                or (is_request_trace and historic_trace.request == subject)
                or (is_response_trace and historic_trace.response == subject)):
                trace = historic_trace
        if trace is None:
            raise AthenaException(f"unable to resolve parent for trace {subject}")

        self.__history_lookup_cache[subject] = trace
        return trace

client(base_build_request=None, name=None)

Create a new client.

Parameters:

Name	Type	Description	Default
base_build_request	Callable[[RequestBuilder], [RequestBuilder]] | None	Function to configure all requests sent by the client.	None
name	str | None	Optional name for client.	None

Returns:

Name	Type	Description
Client	Client	The configured client.

Source code in athena/client.py

def client(self, base_build_request: Callable[[RequestBuilder], RequestBuilder] | None=None, name: str | None=None) -> Client:
    """Create a new client.

    Args:
        base_build_request (Callable[[RequestBuilder], [RequestBuilder]] | None): Function to configure all requests sent by the client.
        name (str | None): Optional name for client.

    Returns:
        Client: The configured client.
    """
    return Client(self.__session.session, self.__session.async_session, base_build_request, name, self.__client_pre_hook, self.__client_post_hook)

trace(subject=None)

Get the full AthenaTrace for a given request or response trace.

Parameters:

Name	Type	Description	Default
subject	AthenaTrace | RequestTrace | ResponseTrace | None	The trace to lookup. If no trace is provided, will return the most recent trace.	None

Returns:

Name	Type	Description
AthenaTrace	AthenaTrace	The full AthenaTrace for the subject.

Example:

def run(athena: Athena):
    response = athena.client().get('https://example.com')
    trace = athena.trace(response)
    print(f'request completed in {trace.elapsed} seconds')

Source code in athena/client.py

def trace(self, subject: AthenaTrace | RequestTrace | ResponseTrace | None=None) -> AthenaTrace:
    """Get the full `AthenaTrace` for a given request or response trace.

    Args:
        subject (AthenaTrace | RequestTrace | ResponseTrace | None): The trace to lookup. If no trace is provided, will return the most recent trace.

    Returns:
        AthenaTrace: The full `AthenaTrace` for the subject.

    Example:

        def run(athena: Athena):
            response = athena.client().get('https://example.com')
            trace = athena.trace(response)
            print(f'request completed in {trace.elapsed} seconds')
    """
    traces = self.traces()
    if subject is None:
        if len(traces) == 0:
            raise AthenaException(f"no completed traces in history")
        subject = traces[-1]

    is_trace = isinstance(subject, AthenaTrace)
    is_request_trace = isinstance(subject, RequestTrace)
    is_response_trace = isinstance(subject, ResponseTrace)
    if not (is_trace or is_request_trace or is_response_trace):
        raise AthenaException(f"unable to resolve parent for trace of type {type(subject).__name__}")

    if subject in self.__history_lookup_cache:
        return self.__history_lookup_cache[subject]
    trace = None
    for historic_trace in traces:
        if ((is_trace and historic_trace == subject)
            or (is_request_trace and historic_trace.request == subject)
            or (is_response_trace and historic_trace.response == subject)):
            trace = historic_trace
    if trace is None:
        raise AthenaException(f"unable to resolve parent for trace {subject}")

    self.__history_lookup_cache[subject] = trace
    return trace

traces()

Get all AthenaTraces from the lifetime of this instance.

Returns:

Type	Description
list[AthenaTrace]	list[AthenaTrace]: List of traces.

Source code in athena/client.py

def traces(self) -> list[AthenaTrace]:
    """Get all `AthenaTrace`s from the lifetime of this instance. 

    Returns:
        list[AthenaTrace]: List of traces.
    """
    return [i for i in self.__history if isinstance(i, AthenaTrace)]

Context

Information about the runtime environment of the module

Attributes:

Name	Type	Description
environment	str	name of environment
module_name	str	name of the module
module_path	str	path to the module
root_path	str	path to root of athena directory

Source code in athena/client.py

@serializeable
class Context:
    """Information about the runtime environment of the module

    Attributes:
        environment (str): name of environment
        module_name (str): name of the module
        module_path (str): path to the module
        root_path (str): path to root of athena directory
    """
    def __init__(self,
        environment: str | None,
        module_name: str,
        module_path: str,
        root_path: str
    ):
        self._environment = environment
        self.environment = str(environment)
        self.module_name = module_name
        self.module_path = module_path
        self.root_path = root_path

ResourceFacade

Facade to interact with resource files.

Attributes:

Name	Type	Description
int	TypedResourceFacade	attempt to cast to int when retrieving value
bool	TypedResourceFacade	attempt to cast to bool when retrieving value
str	TypedResourceFacade	attempt to cast to str when retrieving value
float	TypedResourceFacade	attempt to cast to float when retrieving value

Source code in athena/client.py

class ResourceFacade:
    """Facade to interact with resource files.

    Attributes:
        int (TypedResourceFacade): attempt to cast to int when retrieving value
        bool (TypedResourceFacade): attempt to cast to bool when retrieving value
        str (TypedResourceFacade): attempt to cast to str when retrieving value
        float (TypedResourceFacade): attempt to cast to float when retrieving value
    """
    def __init__(self, loader: Callable[[], _resource_type], resource_type: str, environment: str):
        self._loader = loader
        self._resource_type = resource_type
        self._environment = environment
        self.int = TypedResourceFacade(self, lambda x: int(x))
        self.float = TypedResourceFacade(self, lambda x: float(x))
        self.str = TypedResourceFacade(self, lambda x: str(x))
        self.bool = TypedResourceFacade(self, lambda x: bool(x))

    def _load_resource(self, name) -> _resource_value_type:
        success, value = self._try_load_resource(name)
        if success:
            return value

        raise AthenaException(f"unable to find {self._resource_type} \"{name}\" with environment \"{self._environment}\". ensure {self._resource_type}s have at least a default environment.")

    def _try_load_resource(self, name) -> tuple[bool, _resource_value_type]:
        resource = self._loader()
        return try_extract_value_from_resource(resource, name, self._environment)

    def __getitem__(self, key) -> _resource_value_type:
        """Get value by name. Throws an error if the key cannot be found.

        Args:
            key (str): Name of resource to retrieve.

        Returns:
            _resource_value_type: value
        """
        return self._load_resource(key)

    def get(self, key, default: _resource_value_type | None=None) -> _resource_value_type | None:
        """Get value by name. Returns default value if key cannot be found.

        Args:
            key (str): Name of resource to retrieve.
            default (_resource_value_type | None): Default value

        Returns:
            _resource_value_type | None: value
        """
        success, value = self._try_load_resource(key)
        if success:
            return value
        return default

get(key, default=None)

Get value by name. Returns default value if key cannot be found.

Parameters:

Name	Type	Description	Default
key	str	Name of resource to retrieve.	required
default	_resource_value_type | None	Default value	None

Returns:

Type	Description
_resource_value_type | None	_resource_value_type | None: value

Source code in athena/client.py

def get(self, key, default: _resource_value_type | None=None) -> _resource_value_type | None:
    """Get value by name. Returns default value if key cannot be found.

    Args:
        key (str): Name of resource to retrieve.
        default (_resource_value_type | None): Default value

    Returns:
        _resource_value_type | None: value
    """
    success, value = self._try_load_resource(key)
    if success:
        return value
    return default

TypedResourceFacade

Bases: Generic[T]

Typed facade to interact with resource files.

Source code in athena/client.py

class TypedResourceFacade(Generic[T]):
    """Typed facade to interact with resource files.
    """
    def __init__(self, parent: ResourceFacade, caster: Callable[[_resource_value_type], T]):
        self._parent = parent
        self._caster = caster

    def __getitem__(self, key) -> T:
        """Get value by name. Throws an error if the key cannot be found or if the value is the wrong type.

        Args:
            key (str): Name of resource to retrieve.

        Returns:
            _resource_value_type: value
        """
        return self._caster(self._parent[key])

    def get(self, key, default: T | None=None) -> T | None:
        """Get value by name. Returns default value if key cannot be found or value is the wrong type.

        Args:
            key (str): Name of resource to retrieve.
            default (_resource_value_type | None): Default value

        Returns:
            _resource_value_type | None: value
        """
        value = self._parent.get(key)
        if value is not None:
            try:
                return self._caster(value)
            except:
                pass
        return default

get(key, default=None)

Get value by name. Returns default value if key cannot be found or value is the wrong type.

Parameters:

Name	Type	Description	Default
key	str	Name of resource to retrieve.	required
default	_resource_value_type | None	Default value	None

Returns:

Type	Description
T | None	_resource_value_type | None: value

Source code in athena/client.py

def get(self, key, default: T | None=None) -> T | None:
    """Get value by name. Returns default value if key cannot be found or value is the wrong type.

    Args:
        key (str): Name of resource to retrieve.
        default (_resource_value_type | None): Default value

    Returns:
        _resource_value_type | None: value
    """
    value = self._parent.get(key)
    if value is not None:
        try:
            return self._caster(value)
        except:
            pass
    return default

jsonify(item, *args, **kwargs)

Runs objects through json.dumps, with an encoder for athena objects.

Parameters:

Name	Type	Description	Default
item	Any	The item to json encode.	required

Returns:

Name	Type	Description
str		The json string

Example:

from athena.client import Athena, jsonify

def run(athena: Athena):
    athena.client().get("http://haondt.com")
    traces = athena.traces()
    print(jsonify(traces, indent=4))

Source code in athena/client.py

def jsonify(item: Any, *args, **kwargs):
    """Runs objects through json.dumps, with an encoder for athena objects.

    Args:
        item (Any): The item to json encode.

    Returns:
        str: The json string

    Example:

        from athena.client import Athena, jsonify

        def run(athena: Athena):
            athena.client().get("http://haondt.com")
            traces = athena.traces()
            print(jsonify(traces, indent=4))
    """
    return json_dumps(item, cls=AthenaJSONEncoder, *args, **kwargs)