DataExportClient

armis_sdk.clients.data_export_client.DataExportClient

Bases: BaseEntityClient

Methods:

Name	Description
disable	Disable data export of the entity.
enable	Enable data export of the entity.
iterate	Iterate over the exported data.
get	Get the DataExport of the entity
toggle	Enable / disable export of an entity.

disable(entity) async

Disable data export of the entity.

Parameters:

Name	Type	Description	Default
entity	type[BaseExportedEntity]	The entity to disable exporting for.	required

Example

import asyncio

from armis_sdk.clients.data_export_client import DataExportClient
from armis_sdk.entities.data_export.application import Application


async def main():
    data_export_client = DataExportClient()
    await data_export_client.disable(Application)


asyncio.run(main())

enable(entity) async

Enable data export of the entity.

Parameters:

Name	Type	Description	Default
entity	type[BaseExportedEntity]	The entity to enable exporting for.	required

Example

import asyncio

from armis_sdk.clients.data_export_client import DataExportClient
from armis_sdk.entities.data_export.application import Application


async def main():
    data_export_client = DataExportClient()
    await data_export_client.enable(Application)


asyncio.run(main())

iterate(entity, **kwargs) async

Iterate over the exported data.

Parameters:

Name	Type	Description	Default
entity	type[T]	The entity type to iterate over (must be a subclass of BaseExportedEntity).	required
**kwargs	Any	Additional keyword arguments to pass to pandas.read_parquet().	{}

Returns:

Type	Description
AsyncIterator[T]	An (async) iterator of the underlying entity.

Raises:

Type	Description
ArmisError	If data export is disabled for the entity or if the file format is not parquet.

Example

import asyncio

from armis_sdk.clients.data_export_client import DataExportClient
from armis_sdk.entities.data_export.application import Application


async def main():
    data_export_client = DataExportClient()
    async for row in data_export_client.iterate(Application):
        print(type(row))


asyncio.run(main())

Will output:

<class 'armis_sdk.entities.data_export.application.Application'>
<class 'armis_sdk.entities.data_export.application.Application'>
<class 'armis_sdk.entities.data_export.application.Application'>

Example

You can also pass additional parquet kwargs to filter columns or apply other parquet-specific operations:

import asyncio

from armis_sdk.clients.data_export_client import DataExportClient
from armis_sdk.entities.data_export.application import Application


async def main():
    data_export_client = DataExportClient()
    async for row in data_export_client.iterate(
        Application,
        columns=["device_id", "vendor", "name", "version"],
        filters=[("vendor", "in", ["Google", "Microsoft"])],
    ):
        print(row.device_id, row.vendor, row.name, row.version)


asyncio.run(main())

get(entity) async

Get the DataExport of the entity

Parameters:

Name	Type	Description	Default
entity	type[BaseExportedEntity]	The entity to get the data for.	required

Returns:

Type	Description
DataExport	A DataExport object.

Example

import asyncio

from armis_sdk.clients.data_export_client import DataExportClient
from armis_sdk.entities.data_export.application import Application


async def main():
    data_export_client = DataExportClient()
    print(await data_export_client.get(Application))


asyncio.run(main())

Will output:

DataExport(...)

toggle(entity, enabled) async

Enable / disable export of an entity.

Parameters:

Name	Type	Description	Default
entity	type[BaseExportedEntity]	The entity to enable/disable exporting for.	required
enabled	bool	The new value to set.	required

Raises:

Type	Description
ResponseError	If an error occurs while communicating with the API.

Example

import asyncio

from armis_sdk.clients.data_export_client import DataExportClient
from armis_sdk.entities.data_export.application import Application


async def main():
    data_export_client = DataExportClient()
    await data_export_client.toggle(Application, True)


asyncio.run(main())