Source code for pyEX.rules

# *****************************************************************************
#
# Copyright (c) 2020, the pyEX authors.
#
# This file is part of the pyEX library, distributed under the terms of
# the Apache License 2.0.  The full license can be found in the LICENSE file.
#
from functools import wraps

from ..common import PyEXception, _delete, _get, _post, _raiseIfNotStr
from .engine import Rule  # noqa: F401


def lookup(lookup="", token="", version="stable", format="json"):
    """Pull the latest schema for data points, notification types, and operators used to construct rules.

    https://iexcloud.io/docs/api/#rules-schema

    Args:
        lookup (str): If a schema object has “isLookup”: true, pass the value key to /stable/rules/lookup/{value}. This returns all valid values for the rightValue of a condition.
        token (str): Access token
        version (str): API version
        format (str): return format, defaults to json

    Returns:
        dict: result
    """
    _raiseIfNotStr(lookup)
    if lookup:
        return _get(
            "rules/lookup/{}".format(lookup),
            token=token,
            version=version,
            format=format,
        )
    return _get("rules/schema", token=token, version=version, format=format)


@wraps(lookup)
def schema(token="", version="stable", format="json"):
    return lookup(token=token, version=version, format=format)


def create(
    rule,
    ruleName,
    ruleSet,
    type="any",
    existingId=None,
    token="",
    version="stable",
    format="json",
):
    """This endpoint is used to both create and edit rules. Note that rules run be default after being created.

    Args:
        rule (Rule or dict): rule object to create
        ruleName (str): name for rule
        ruleSet (str): Valid US symbol or the string ANYEVENT. If the string ANYEVENT is passed, the rule will be triggered for any symbol in the system. The cool down period for alerts (frequency) is applied on a per symbol basis.
        type (str): Specify either any, where if any condition is true you get an alert, or all, where all conditions must be true to trigger an alert. any is the default value
        existingId (Optional[str]): The id of an existing rule only if you are editing the existing rule
        token (str): Access token
        version (str): API version
        format (str): return format, defaults to json
    """
    if type not in ("any", "all"):
        raise PyEXception("type must be in (any, all). got: {}".format(type))

    if isinstance(rule, Rule):
        rule = rule.toJson()

    rule["token"] = token
    rule["ruleSet"] = ruleSet
    rule["type"] = type
    rule["ruleName"] = ruleName

    # Conditions, outputs, and additionalKeys handled by rule object
    if "conditions" not in rule:
        raise PyEXception("rule is missing `conditions` key!")
    if "outputs" not in rule:
        raise PyEXception("rule is missing `outputs` key!")

    if existingId is not None:
        rule["id"] = existingId
    return _post(
        "rules/create",
        json=rule,
        token=token,
        version=version,
        token_in_params=False,
        format=format,
    )


def pause(ruleId, token="", version="stable", format="json"):
    """You can control the output of rules by pausing and resume per rule id.

    Args:
        ruleId (str): The id of an existing rule to puase
        token (str): Access token
        version (str): API version
        format (str): return format, defaults to json
    """
    return _post(
        "rules/pause",
        json={"ruleId": ruleId, "token": token},
        token=token,
        version=version,
        token_in_params=False,
        format=format,
    )


def resume(ruleId, token="", version="stable", format="json"):
    """You can control the output of rules by pausing and resume per rule id.

    Args:
        ruleId (str): The id of an existing rule to puase
        token (str): Access token
        version (str): API version
        format (str): return format, defaults to json
    """
    return _post(
        "rules/resume",
        json={"ruleId": ruleId, "token": token},
        token=token,
        version=version,
        token_in_params=False,
        format=format,
    )


def delete(ruleId, token="", version="stable", format="json"):
    """You can delete a rule by using an __HTTP DELETE__ request. This will stop rule executions and delete the rule from your dashboard. If you only want to temporarily stop a rule, use the pause/resume functionality instead.

    Args:
        ruleId (str): The id of an existing rule to puase
        token (str): Access token
        version (str): API version
        format (str): return format, defaults to json
    """
    return _delete(
        "rules/{}".format(ruleId), token=token, version=version, format=format
    )


def rule(ruleId, token="", version="stable", format="json"):
    """Rule information such as the current rule status and execution statistics.

    Args:
        ruleId (str): The id of an existing rule to puase
        token (str): Access token
        version (str): API version
        format (str): return format, defaults to json
    """
    return _get(
        "rules/info/{}".format(ruleId), token=token, version=version, format=format
    )


def rules(token="", version="stable", format="json"):
    """List all rules that are currently on your account. Each rule object returned will include the current rule status and execution statistics."""
    return _get("rules", token=token, version=version, format=format)


def output(ruleId, token="", version="stable", format="json"):
    """If you choose `logs` as your rule output method, IEX Cloud will save the output objects on our server. You can use this method to retrieve those data objects.

    Args:
        ruleId (str): The id of an existing rule to puase
        token (str): Access token
        version (str): API version
        format (str): return format, defaults to json
    """
    return _get(
        "rules/output/{}".format(ruleId), token=token, version=version, format=format
    )