Manager

psengine.risklists.risklist_mgr.RisklistMgr

RisklistMgr(rf_token: Optional[str] = None)

Manages requests for Recorded Future risk lists.

PARAMETER	DESCRIPTION
rf_token	Recorded Future API token. TYPE: Optional[str] DEFAULT: None

Source code in psengine/risklists/risklist_mgr.py

def __init__(
    self,
    rf_token: Annotated[Optional[str], Doc('Recorded Future API token.')] = None,
):
    """Initializes the RiskListMgr object."""
    self.log = logging.getLogger(__name__)
    self.rf_client = RFClient(api_token=rf_token) if rf_token else RFClient()

fetch_risklist

fetch_risklist(
    list: str,
    entity_type: Optional[str] = None,
    format: Optional[str] = None,
    headers: bool = True,
    validate: Optional[Any] = None,
) -> Generator[
    Union[dict, list[str], BaseModel], None, None
]

Get a Recorded Future RiskList as generator.

For a custom risklist, specify a fusion_path the format field is ignored when custom risklists are used.

PARAMETER	DESCRIPTION
list	Name of the risklist to download. TYPE: str
entity_type	Type of entity to get risklist for. TYPE: Optional[str] DEFAULT: None
format	Format of the risklist. TYPE: Optional[str] DEFAULT: None
headers	Whether headers are included in the CSV. TYPE: bool DEFAULT: True
validate	Validation model to use. Must be a subclass of pydantic BaseModel. TYPE: Optional[Any] DEFAULT: None

Warning

• If a specified list does't exist, the API returns the default risklist.
• An empty risklist may be returned:
  • If validate is None and headers are included, headers are returned.
  • If validate is set, an empty list is returned.

Example

Download and return entries as JSON:

from psengine.risklists import RisklistMgr, DefaultRiskList

mgr = RisklistMgr()
data = mgr.fetch_risklist('default', 'domain', validate=DefaultRiskList)
for entry in data:
    print(entry.json())

RAISES	DESCRIPTION
RisklistNotAvailableError	If an HTTP error occurs during risklist fetch.
ValidationError	If any parameter is of incorrect type.

RETURNS	DESCRIPTION
Generator[Union[dict, list[str], BaseModel], None, None]	Yields risklist rows or validated risklist models.

Source code in psengine/risklists/risklist_mgr.py

@debug_call
@validate_call
def fetch_risklist(
    self,
    list: Annotated[str, Doc('Name of the risklist to download.')],  # noqa: A002
    entity_type: Annotated[Optional[str], Doc('Type of entity to get risklist for.')] = None,
    format: Annotated[Optional[str], Doc('Format of the risklist.')] = None,  # noqa: A002
    headers: Annotated[bool, Doc('Whether headers are included in the CSV.')] = True,
    validate: Annotated[
        Optional[Any], Doc('Validation model to use. Must be a subclass of pydantic BaseModel.')
    ] = None,
) -> Annotated[
    Generator[Union[dict, list[str], BaseModel], None, None],
    Doc('Yields risklist rows or validated risklist models.'),
]:
    """Get a Recorded Future RiskList as generator.

    For a custom risklist, specify a `fusion_path` the `format` field is ignored
    when custom risklists are used.

    Warning:
        - If a specified list does't exist, the API returns the default risklist.
        - An empty risklist may be returned:
            - If `validate` is None and headers are included, headers are returned.
            - If `validate` is set, an empty list is returned.

    Example:
        Download and return entries as JSON:

        ```python
        from psengine.risklists import RisklistMgr, DefaultRiskList

        mgr = RisklistMgr()
        data = mgr.fetch_risklist('default', 'domain', validate=DefaultRiskList)
        for entry in data:
            print(entry.json())
        ```

    Raises:
        RisklistNotAvailableError: If an HTTP error occurs during risklist fetch.
        ValidationError: If any parameter is of incorrect type.
    """
    if validate and not issubclass(validate, BaseModel):
        raise ValueError('`validate` should be a subclass of Pydantic BaseModel or None')

    format = format or DEFAULT_RISKLIST_FORMAT  # noqa: A001
    risklist_type, url, params = self._get_risklist_url_and_params(list, entity_type, format)

    if risklist_type == 'fusion' and list.endswith('json'):
        return self._fetch_json_risklist(url, params, validate)
    return self._fetch_csv_risklist(url, params, validate, headers)