Skip to content

ADT

psengine.entity_lists.entity_list

EntityList

Bases: RFBaseModel

Validate data received from /create endpoint.

created instance-attribute 

created: datetime

id_ class-attribute instance-attribute 

id_: str = Field(alias='id')

log class-attribute instance-attribute 

log: Logger = Field(
    exclude=True, default=getLogger(__name__)
)

match_mgr class-attribute instance-attribute 

match_mgr: EntityMatchMgr = Field(exclude=True)

model_config class-attribute instance-attribute 

model_config = ConfigDict(arbitrary_types_allowed=True)

name instance-attribute 

name: str

organisation_id class-attribute instance-attribute 

organisation_id: Optional[str] = None

organisation_name class-attribute instance-attribute 

organisation_name: Optional[str] = None

owner_id instance-attribute 

owner_id: str

owner_name instance-attribute 

owner_name: str

owner_organisation_details class-attribute instance-attribute 

owner_organisation_details: OwnerOrganisationDetails = (
    Field(default_factory=OwnerOrganisationDetails)
)

rf_client class-attribute instance-attribute 

rf_client: RFClient = Field(exclude=True)

type_ class-attribute instance-attribute 

type_: str = Field(alias='type')

updated instance-attribute 

updated: datetime

__eq__ 

__eq__(other: EntityList)
Source code in psengine/entity_lists/entity_list.py 
111
112
def __eq__(self, other: 'EntityList'):
    return self.id_ == other.id_

__hash__ 

__hash__()
Source code in psengine/entity_lists/entity_list.py 
108
109
def __hash__(self):
    return hash(self.id_)

__str__ 

__str__() -> str

Return the string representation of the list.

RETURNS DESCRIPTION
str

List data with standard info and entities.
Source code in psengine/entity_lists/entity_list.py 
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
def __str__(self) -> Annotated[str, Doc('List data with standard info and entities.')]:
    """Return the string representation of the list."""

    def format_date(date):
        return date.strftime(TIMESTAMP_STR)

    def format_field(name, value):
        return f'{name}: {value or "None"}'

    main_fields = [
        format_field('id', self.id_),
        format_field('name', self.name),
        format_field('type', self.type_),
        format_field('created', format_date(self.created)),
        format_field('last updated', format_date(self.updated)),
        format_field('owner id', self.owner_id),
        format_field('owner name', self.owner_name),
        format_field('organisation id', self.organisation_id),
        format_field('organisation name', self.organisation_name),
    ]

    org_details = self.owner_organisation_details
    org_fields = [
        format_field('owner id', org_details.owner_id),
        format_field('owner name', org_details.owner_name),
        format_field('enterprise id', org_details.enterprise_id),
        format_field('enterprise name', org_details.enterprise_name),
    ]

    sub_orgs = org_details.organisations
    if sub_orgs:
        sub_org_str = '\n    '.join(
            f'organisation id: {org.organisation_id}\n'
            f'    organisation name: {org.organisation_name}'
            for org in sub_orgs
        )
        org_fields.append(f'sub-organisations:\n    {sub_org_str}')
    else:
        org_fields.append('sub-organisations: None')

    return (
        '\n'.join(main_fields) + '\nowner organisation details:\n  ' + '\n  '.join(org_fields)
    )

add 

add(
    entity: Union[str, tuple[str, str]],
    context: Optional[dict] = None,
) -> ListEntityOperationResponse

Add an entity to a list.

PARAMETER DESCRIPTION
entity

ID or (name, type) tuple of the entity to add.

TYPE: Union[str, tuple[str, str]]

context

Context object for the entity.

TYPE: Optional[dict] DEFAULT: None

Endpoint

list/{id}/entity/add

RAISES DESCRIPTION
ValidationError

if any supplied parameter is of incorrect type.
ListApiError

If connection error occurs.
RETURNS DESCRIPTION
ListEntityOperationResponse

Response from the list/{id}/entity/add endpoint.
Source code in psengine/entity_lists/entity_list.py 
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
@debug_call
@validate_call
def add(
    self,
    entity: Annotated[
        Union[str, tuple[str, str]], Doc('ID or (name, type) tuple of the entity to add.')
    ],
    context: Annotated[Optional[dict], Doc('Context object for the entity.')] = None,
) -> Annotated[
    ListEntityOperationResponse, Doc('Response from the `list/{id}/entity/add` endpoint.')
]:
    """Add an entity to a list.

    Endpoint:
        `list/{id}/entity/add`

    Raises:
        ValidationError: if any supplied parameter is of incorrect type.
        ListApiError: If connection error occurs.
    """
    return self._list_op(entity, ADD_OP, context=context or {})

bulk_add 

bulk_add(
    entities: list[Union[str, tuple[str, str]]],
) -> dict

Bulk add entities to a list.

Adds entities one at a time due to List API requirement. Logs progress every 10%.

PARAMETER DESCRIPTION
entities

List of entity string IDs or (name, type) tuples to add.

TYPE: list[Union[str, tuple[str, str]]]

Endpoint

list/{id}/entity/add

RAISES DESCRIPTION
ValidationError

If any supplied parameter is of incorrect type.
ValueError

If an invalid operation is supplied.
ListApiError

If connection error occurs.
RETURNS DESCRIPTION
dict

Results JSON with 'added', 'unchanged', and 'error' keys containing lists of entities.
Source code in psengine/entity_lists/entity_list.py 
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
@debug_call
@validate_call
def bulk_add(
    self,
    entities: Annotated[
        list[Union[str, tuple[str, str]]],
        Doc('List of entity string IDs or (name, type) tuples to add.'),
    ],
) -> Annotated[
    dict,
    Doc(
        "Results JSON with 'added', 'unchanged', and 'error' keys containing lists of entities."
    ),
]:
    """Bulk add entities to a list.

    Adds entities one at a time due to List API requirement. Logs progress every 10%.

    Endpoint:
        `list/{id}/entity/add`

    Raises:
        ValidationError: If any supplied parameter is of incorrect type.
        ValueError: If an invalid operation is supplied.
        ListApiError: If connection error occurs.
    """
    result = self._bulk_op(entities, ADD_OP)
    status = self.status()
    while status.status != 'ready':
        self.log.info(f"Awaiting list 'ready' status, current status '{status.status}'")
        status = self.status()
        time.sleep(IS_READY_INCREMENT)

    return result

bulk_remove 

bulk_remove(
    entities: list[Union[str, tuple[str, str]]],
) -> dict

Bulk remove entities from a list.

Removes entities one at a time due to List API requirement. Logs progress every 10%.

PARAMETER DESCRIPTION
entities

List of entity string IDs or (name, type) tuples to remove.

TYPE: list[Union[str, tuple[str, str]]]

Endpoint

list/{id}/entity/remove

RAISES DESCRIPTION
ValidationError

If any supplied parameter is of incorrect type.
ValueError

If an invalid operation is supplied.
ListApiError

If connection error occurs.
RETURNS DESCRIPTION
dict

Results JSON with 'removed', 'unchanged', and 'error' keys with the lists of entities.
Source code in psengine/entity_lists/entity_list.py 
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
@debug_call
@validate_call
def bulk_remove(
    self,
    entities: Annotated[
        list[Union[str, tuple[str, str]]],
        Doc('List of entity string IDs or (name, type) tuples to remove.'),
    ],
) -> Annotated[
    dict,
    Doc(
        "Results JSON with 'removed', 'unchanged', and 'error' keys with the lists of entities."
    ),
]:
    """Bulk remove entities from a list.

    Removes entities one at a time due to List API requirement. Logs progress every 10%.

    Endpoint:
        `list/{id}/entity/remove`

    Raises:
        ValidationError: If any supplied parameter is of incorrect type.
        ValueError: If an invalid operation is supplied.
        ListApiError: If connection error occurs.
    """
    result = self._bulk_op(entities, REMOVE_OP)
    status = self.status()
    while status.status != 'ready':
        self.log.info(f"Awaiting list 'ready' status, current status '{status.status}'")
        status = self.status()
        time.sleep(IS_READY_INCREMENT)

    return result

entities 

entities() -> list[ListEntity]

Get entities for a list.

Endpoint

list/{id}/entities

RAISES DESCRIPTION
ListApiError

If connection error occurs.
RETURNS DESCRIPTION
list[ListEntity]

Response from the list/{id}/entities endpoint.
Source code in psengine/entity_lists/entity_list.py 
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
@debug_call
@connection_exceptions(ignore_status_code=[], exception_to_raise=ListApiError)
def entities(
    self,
) -> Annotated[list[ListEntity], Doc('Response from the `list/{id}/entities` endpoint.')]:
    """Get entities for a list.

    Endpoint:
        `list/{id}/entities`

    Raises:
        ListApiError: If connection error occurs.
    """
    url = EP_LIST + '/' + self.id_ + '/entities'
    response = self.rf_client.request('get', url)
    return [ListEntity.model_validate(entity) for entity in response.json()]

info 

info() -> ListInfoOut

Get info for a list.

Endpoint

list/{id}/info

RAISES DESCRIPTION
ListApiError

If connection error occurs.
RETURNS DESCRIPTION
ListInfoOut

Response from the list/{id}/info endpoint.
Source code in psengine/entity_lists/entity_list.py 
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
@debug_call
@connection_exceptions(ignore_status_code=[], exception_to_raise=ListApiError)
def info(self) -> Annotated[ListInfoOut, Doc('Response from the `list/{id}/info` endpoint.')]:
    """Get info for a list.

    Endpoint:
        `list/{id}/info`

    Raises:
        ListApiError: If connection error occurs.
    """
    self.log.debug(f"Getting list status for '{self.name}'")
    url = EP_LIST + f'/{self.id_}/info'
    response = self.rf_client.request('get', url)
    return ListInfoOut.model_validate(response.json())

json 

json(
    by_alias: bool = True,
    exclude_none: bool = True,
    auto_exclude_unset: bool = True,
    **kwargs,
)

JSON representation of models. It is inherited by every model.

PARAMETER DESCRIPTION
by_alias

Alias flag:

  • If True, writes fields with their API alias (e.g., IpAddress)
  • If False uses the Python attribute name alias.

TYPE: bool DEFAULT: True

exclude_none

Whether to exclude fields equal to None.

TYPE: bool DEFAULT: True

auto_exclude_unset

Whether to auto exclude values not set.

  • If True, uses RF_EXTRA_MODEL config to decide inclusion of unmapped fields.
  • If False, you must specify exclude_unset manually.

TYPE: bool DEFAULT: True

Source code in psengine/common_models.py 
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
def json(
    self,
    by_alias: Annotated[
        bool,
        Doc(
            """
            Alias flag:

            - If `True`, writes fields with their API alias (e.g., `IpAddress`)
            - If `False` uses the Python attribute name alias.
            """
        ),
    ] = True,
    exclude_none: Annotated[bool, Doc('Whether to exclude fields equal to None.')] = True,
    auto_exclude_unset: Annotated[
        bool,
        Doc("""
            Whether to auto exclude values not set.

            - If `True`, uses `RF_EXTRA_MODEL` config to decide inclusion of unmapped fields.
            - If `False`, you must specify `exclude_unset` manually.
            """),
    ] = True,
    **kwargs,
):
    """JSON representation of models. It is inherited by every model."""
    if not auto_exclude_unset and kwargs.get('exclude_unset') is None:
        raise ValueError('`auto_exclude_unset` is False, `exclude_unset has to be provided`')

    exclude_unset = (
        bool(self.model_config['extra'] != 'allow')
        if auto_exclude_unset
        else kwargs['exclude_unset']
    )
    kwargs['exclude_unset'] = exclude_unset
    return self.model_dump(mode='json', by_alias=by_alias, exclude_none=exclude_none, **kwargs)

remove 

remove(
    entity: Union[str, tuple[str, str]],
) -> ListEntityOperationResponse

Remove an entity from a list.

PARAMETER DESCRIPTION
entity

ID or (name, type) tuple of the entity to remove.

TYPE: Union[str, tuple[str, str]]

Endpoint

list/{id}/entity/remove

RAISES DESCRIPTION
ValidationError

if any supplied parameter is of incorrect type.
ListApiError

If connection error occurs.
RETURNS DESCRIPTION
ListEntityOperationResponse

Response from the list/{id}/entity/remove endpoint.
Source code in psengine/entity_lists/entity_list.py 
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
@debug_call
@validate_call
def remove(
    self,
    entity: Annotated[
        Union[str, tuple[str, str]], Doc('ID or (name, type) tuple of the entity to remove.')
    ],
) -> Annotated[
    ListEntityOperationResponse, Doc('Response from the `list/{id}/entity/remove` endpoint.')
]:
    """Remove an entity from a list.

    Endpoint:
        `list/{id}/entity/remove`

    Raises:
        ValidationError: if any supplied parameter is of incorrect type.
        ListApiError: If connection error occurs.
    """
    return self._list_op(entity, REMOVE_OP)

status 

status() -> ListStatusOut

Get status information about list.

Endpoint

list/{id}/status

RAISES DESCRIPTION
ListApiError

if connection error occurs
RETURNS DESCRIPTION
ListStatusOut

list/{id}/status response

TYPE: ListStatusOut

Source code in psengine/entity_lists/entity_list.py 
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
@debug_call
@connection_exceptions(ignore_status_code=[], exception_to_raise=ListApiError)
def status(self) -> ListStatusOut:
    """Get status information about list.

    Endpoint:
        `list/{id}/status`

    Raises:
        ListApiError: if connection error occurs

    Returns:
        ListStatusOut: list/{id}/status response
    """
    self.log.debug(f"Getting list status for '{self.name}'")
    url = EP_LIST + f'/{self.id_}/status'
    response = self.rf_client.request('get', url)
    validated_status = ListStatusOut.model_validate(response.json())
    self.log.debug(
        f"List '{self.name}' status: {validated_status.status}, "
        f'entities: {validated_status.size}'
    )

    return validated_status

text_entries 

text_entries() -> list[str]

Get text entries for a list.

Endpoint

list/{id}/textEntries

RAISES DESCRIPTION
ListApiError

If connection error occurs.
RETURNS DESCRIPTION
list[str]

Response from the list/{id}/textEntries endpoint.
Source code in psengine/entity_lists/entity_list.py 
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
@debug_call
@connection_exceptions(ignore_status_code=[], exception_to_raise=ListApiError)
def text_entries(
    self,
) -> Annotated[list[str], Doc('Response from the `list/{id}/textEntries` endpoint.')]:
    """Get text entries for a list.

    Endpoint:
        `list/{id}/textEntries`

    Raises:
        ListApiError: If connection error occurs.
    """
    url = EP_LIST + '/' + self.id_ + '/textEntries'
    return self.rf_client.request('get', url).json()

ListEntity

Bases: RFBaseModel

Validate data received from /{listId}/entities endpoint.

added instance-attribute 

added: datetime

context class-attribute instance-attribute 

context: Optional[dict] = None

entity instance-attribute 

entity: IdNameType

model_config class-attribute instance-attribute 

model_config = ConfigDict(
    extra=get('RF_MODEL_EXTRA', 'ignore')
)

status instance-attribute 

status: str

__eq__ 

__eq__(other: ListEntity)
Source code in psengine/entity_lists/entity_list.py 
76
77
def __eq__(self, other: 'ListEntity'):
    return self.entity.id_ == other.entity.id_

__gt__ 

__gt__(other: ListEntity)
Source code in psengine/entity_lists/entity_list.py 
79
80
def __gt__(self, other: 'ListEntity'):
    return (self.entity.name, self.added) > (other.entity.name, other.added)

__hash__ 

__hash__()
Source code in psengine/entity_lists/entity_list.py 
73
74
def __hash__(self):
    return hash(self.entity.id_)

__str__ 

__str__()
Source code in psengine/entity_lists/entity_list.py 
82
83
84
85
def __str__(self):
    return (
        f'{self.entity.type_}: {self.entity.name}, added {self.added.strftime(TIMESTAMP_STR)}'
    )

json 

json(
    by_alias: bool = True,
    exclude_none: bool = True,
    auto_exclude_unset: bool = True,
    **kwargs,
)

JSON representation of models. It is inherited by every model.

PARAMETER DESCRIPTION
by_alias

Alias flag:

  • If True, writes fields with their API alias (e.g., IpAddress)
  • If False uses the Python attribute name alias.

TYPE: bool DEFAULT: True

exclude_none

Whether to exclude fields equal to None.

TYPE: bool DEFAULT: True

auto_exclude_unset

Whether to auto exclude values not set.

  • If True, uses RF_EXTRA_MODEL config to decide inclusion of unmapped fields.
  • If False, you must specify exclude_unset manually.

TYPE: bool DEFAULT: True

Source code in psengine/common_models.py 
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
def json(
    self,
    by_alias: Annotated[
        bool,
        Doc(
            """
            Alias flag:

            - If `True`, writes fields with their API alias (e.g., `IpAddress`)
            - If `False` uses the Python attribute name alias.
            """
        ),
    ] = True,
    exclude_none: Annotated[bool, Doc('Whether to exclude fields equal to None.')] = True,
    auto_exclude_unset: Annotated[
        bool,
        Doc("""
            Whether to auto exclude values not set.

            - If `True`, uses `RF_EXTRA_MODEL` config to decide inclusion of unmapped fields.
            - If `False`, you must specify `exclude_unset` manually.
            """),
    ] = True,
    **kwargs,
):
    """JSON representation of models. It is inherited by every model."""
    if not auto_exclude_unset and kwargs.get('exclude_unset') is None:
        raise ValueError('`auto_exclude_unset` is False, `exclude_unset has to be provided`')

    exclude_unset = (
        bool(self.model_config['extra'] != 'allow')
        if auto_exclude_unset
        else kwargs['exclude_unset']
    )
    kwargs['exclude_unset'] = exclude_unset
    return self.model_dump(mode='json', by_alias=by_alias, exclude_none=exclude_none, **kwargs)

ListInfoOut

Bases: RFBaseModel

Validate data received from /{listId}/info endpoint.

created instance-attribute 

created: datetime

id_ class-attribute instance-attribute 

id_: str = Field(alias='id')

model_config class-attribute instance-attribute 

model_config = ConfigDict(
    extra=get('RF_MODEL_EXTRA', 'ignore')
)

name instance-attribute 

name: str

organisation_id instance-attribute 

organisation_id: str

organisation_name instance-attribute 

organisation_name: str

owner_id instance-attribute 

owner_id: str

owner_name instance-attribute 

owner_name: str

owner_organisation_details class-attribute instance-attribute 

owner_organisation_details: OwnerOrganisationDetails = (
    Field(default_factory=OwnerOrganisationDetails)
)

type_ class-attribute instance-attribute 

type_: str = Field(alias='type')

updated instance-attribute 

updated: datetime

json 

json(
    by_alias: bool = True,
    exclude_none: bool = True,
    auto_exclude_unset: bool = True,
    **kwargs,
)

JSON representation of models. It is inherited by every model.

PARAMETER DESCRIPTION
by_alias

Alias flag:

  • If True, writes fields with their API alias (e.g., IpAddress)
  • If False uses the Python attribute name alias.

TYPE: bool DEFAULT: True

exclude_none

Whether to exclude fields equal to None.

TYPE: bool DEFAULT: True

auto_exclude_unset

Whether to auto exclude values not set.

  • If True, uses RF_EXTRA_MODEL config to decide inclusion of unmapped fields.
  • If False, you must specify exclude_unset manually.

TYPE: bool DEFAULT: True

Source code in psengine/common_models.py 
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
def json(
    self,
    by_alias: Annotated[
        bool,
        Doc(
            """
            Alias flag:

            - If `True`, writes fields with their API alias (e.g., `IpAddress`)
            - If `False` uses the Python attribute name alias.
            """
        ),
    ] = True,
    exclude_none: Annotated[bool, Doc('Whether to exclude fields equal to None.')] = True,
    auto_exclude_unset: Annotated[
        bool,
        Doc("""
            Whether to auto exclude values not set.

            - If `True`, uses `RF_EXTRA_MODEL` config to decide inclusion of unmapped fields.
            - If `False`, you must specify `exclude_unset` manually.
            """),
    ] = True,
    **kwargs,
):
    """JSON representation of models. It is inherited by every model."""
    if not auto_exclude_unset and kwargs.get('exclude_unset') is None:
        raise ValueError('`auto_exclude_unset` is False, `exclude_unset has to be provided`')

    exclude_unset = (
        bool(self.model_config['extra'] != 'allow')
        if auto_exclude_unset
        else kwargs['exclude_unset']
    )
    kwargs['exclude_unset'] = exclude_unset
    return self.model_dump(mode='json', by_alias=by_alias, exclude_none=exclude_none, **kwargs)

ListStatusOut

Bases: RFBaseModel

Validate data received from /{listId}/status endpoint.

model_config class-attribute instance-attribute 

model_config = ConfigDict(
    extra=get('RF_MODEL_EXTRA', 'ignore')
)

size instance-attribute 

size: int

status instance-attribute 

status: str

json 

json(
    by_alias: bool = True,
    exclude_none: bool = True,
    auto_exclude_unset: bool = True,
    **kwargs,
)

JSON representation of models. It is inherited by every model.

PARAMETER DESCRIPTION
by_alias

Alias flag:

  • If True, writes fields with their API alias (e.g., IpAddress)
  • If False uses the Python attribute name alias.

TYPE: bool DEFAULT: True

exclude_none

Whether to exclude fields equal to None.

TYPE: bool DEFAULT: True

auto_exclude_unset

Whether to auto exclude values not set.

  • If True, uses RF_EXTRA_MODEL config to decide inclusion of unmapped fields.
  • If False, you must specify exclude_unset manually.

TYPE: bool DEFAULT: True

Source code in psengine/common_models.py 
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
def json(
    self,
    by_alias: Annotated[
        bool,
        Doc(
            """
            Alias flag:

            - If `True`, writes fields with their API alias (e.g., `IpAddress`)
            - If `False` uses the Python attribute name alias.
            """
        ),
    ] = True,
    exclude_none: Annotated[bool, Doc('Whether to exclude fields equal to None.')] = True,
    auto_exclude_unset: Annotated[
        bool,
        Doc("""
            Whether to auto exclude values not set.

            - If `True`, uses `RF_EXTRA_MODEL` config to decide inclusion of unmapped fields.
            - If `False`, you must specify `exclude_unset` manually.
            """),
    ] = True,
    **kwargs,
):
    """JSON representation of models. It is inherited by every model."""
    if not auto_exclude_unset and kwargs.get('exclude_unset') is None:
        raise ValueError('`auto_exclude_unset` is False, `exclude_unset has to be provided`')

    exclude_unset = (
        bool(self.model_config['extra'] != 'allow')
        if auto_exclude_unset
        else kwargs['exclude_unset']
    )
    kwargs['exclude_unset'] = exclude_unset
    return self.model_dump(mode='json', by_alias=by_alias, exclude_none=exclude_none, **kwargs)