Manager

psengine.entity_lists.entity_list_mgr.EntityListMgr

EntityListMgr(rf_token: Optional[str] = None)

Manages requests for Recorded Future List API.

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

Source code in psengine/entity_lists/entity_list_mgr.py

def __init__(
    self,
    rf_token: Annotated[Optional[str], Doc('Recorded Future API token.')] = None,
) -> None:
    """Initialize the `EntityListMgr` object."""
    self.log = logging.getLogger(__name__)
    self.rf_client = RFClient(api_token=rf_token) if rf_token else RFClient()
    self.match_mgr = EntityMatchMgr(rf_token=rf_token) if rf_token else EntityMatchMgr()

create

create(
    list_name: str, list_type: str = 'entity'
) -> EntityList

Create a new list.

PARAMETER	DESCRIPTION
list_name	List name to use for the new list. TYPE: str
list_type	List type. Supported types are documented in the List API support page. TYPE: str DEFAULT: 'entity'

Endpoint

list/create

RAISES	DESCRIPTION
ValidationError	If any supplied parameter is of incorrect type.
ListApiError	If connection error occurs.

RETURNS	DESCRIPTION
EntityList	EntityList object for the new list.

Source code in psengine/entity_lists/entity_list_mgr.py

@debug_call
@validate_call
@connection_exceptions(ignore_status_code=[], exception_to_raise=ListApiError)
def create(
    self,
    list_name: Annotated[str, Doc('List name to use for the new list.')],
    list_type: Annotated[
        str, Doc('List type. Supported types are documented in the List API support page.')
    ] = 'entity',
) -> Annotated[EntityList, Doc('EntityList object for the new list.')]:
    """Create a new list.

    Endpoint:
        `list/create`

    Raises:
        ValidationError: If any supplied parameter is of incorrect type.
        ListApiError: If connection error occurs.
    """
    self.log.debug(f"Creating list '{list_name}'")
    request_body = {'name': list_name, 'type': list_type}
    response = self.rf_client.request('post', EP_CREATE_LIST, data=request_body)
    list_create_data = response.json()
    self.log.debug(f"List '{list_name}' created")
    self.log.debug('  ID: {}'.format(list_create_data['id']))
    self.log.debug('  Type: {}'.format(list_create_data['type']))

    return EntityList(rf_client=self.rf_client, match_mgr=self.match_mgr, **list_create_data)

fetch

fetch(list_: Union[str, tuple[str, str]]) -> EntityList

Get a list by its ID. Use this method to retrieve list info.

PARAMETER	DESCRIPTION
list_	List string ID or tuple of (name, type). TYPE: Union[str, tuple[str, str]]

Endpoint

list/{list_id}/info

RAISES	DESCRIPTION
ValidationError	If any supplied parameter is of incorrect type.
ListResolutionError	When list_ is a tuple and name matches zero or multiple entities.
ListApiError	If connection error occurs.

RETURNS	DESCRIPTION
EntityList	RFList object for the given list ID.

Source code in psengine/entity_lists/entity_list_mgr.py

@debug_call
@validate_call
@connection_exceptions(ignore_status_code=[], exception_to_raise=ListApiError)
def fetch(
    self,
    list_: Annotated[
        Union[str, tuple[str, str]], Doc('List string ID or tuple of (name, type).')
    ],
) -> Annotated[EntityList, Doc('RFList object for the given list ID.')]:
    """Get a list by its ID. Use this method to retrieve list info.

    Endpoint:
        `list/{list_id}/info`

    Raises:
        ValidationError: If any supplied parameter is of incorrect type.
        ListResolutionError: When `list_` is a tuple and name matches zero or multiple entities.
        ListApiError: If connection error occurs.
    """
    resolved_id = self._resolve_list_id(list_)
    self.log.info(f'Getting list with ID: {resolved_id}')
    url = EP_LIST + f'/{resolved_id}/info'
    response = self.rf_client.request('get', url)
    list_info_data = response.json()
    self.log.debug("Found list ID '{}'".format(list_info_data['id']))
    self.log.debug('  Type: {}'.format(list_info_data['type']))
    self.log.debug('  Created: {}'.format(list_info_data['created']))
    self.log.debug('  Updated: {}'.format(list_info_data['updated']))

    return EntityList(rf_client=self.rf_client, match_mgr=self.match_mgr, **list_info_data)

search

search(
    list_name: Optional[str] = None,
    list_type: Optional[str] = None,
    max_results: int = DEFAULT_LIMIT,
) -> list[EntityList]

Search lists.

PARAMETER	DESCRIPTION
list_name	List name to search. TYPE: Optional[str] DEFAULT: None
list_type	List type to filter by. Ignored if None. TYPE: Optional[str] DEFAULT: None
max_results	Maximum number of lists to return. TYPE: int DEFAULT: DEFAULT_LIMIT

Endpoint

list/search

RAISES	DESCRIPTION
ValidationError	If any supplied parameter is of incorrect type.
ListApiError	If the list API call fails.

RETURNS	DESCRIPTION
list[EntityList]	List of EntityList objects from list/search.

Source code in psengine/entity_lists/entity_list_mgr.py

@debug_call
@validate_call
@connection_exceptions(ignore_status_code=[], exception_to_raise=ListApiError)
def search(
    self,
    list_name: Annotated[Optional[str], Doc('List name to search.')] = None,
    list_type: Annotated[Optional[str], Doc('List type to filter by. Ignored if None.')] = None,
    max_results: Annotated[int, Doc('Maximum number of lists to return.')] = DEFAULT_LIMIT,
) -> Annotated[list[EntityList], Doc('List of EntityList objects from `list/search`.')]:
    """Search lists.

    Endpoint:
        `list/search`

    Raises:
        ValidationError: If any supplied parameter is of incorrect type.
        ListApiError: If the list API call fails.
    """
    request_body = {}
    request_body['limit'] = max_results
    if list_name:
        request_body['name'] = list_name
    if list_type:
        request_body['type'] = list_type
    self.log.info(f'Searching list API with parameters: {request_body}')
    response = self.rf_client.request('post', EP_SEARCH_LIST, data=request_body)
    list_search_data = response.json()
    self.log.info(
        'Found {} matching {}'.format(
            len(list_search_data), 'lists' if len(list_search_data) != 1 else 'list'
        )
    )

    return [
        EntityList(rf_client=self.rf_client, match_mgr=self.match_mgr, **list_)
        for list_ in list_search_data
    ]