Async File

The async File module provides functionality for working with files.

class AsyncFile(api, uuid, data={})

Bases: AsyncBase

Parameters:

• api (AsyncGeoboxClient)

• uuid (str)

• data (Dict | None)

BASE_ENDPOINT: str = 'files/'

__init__(api, uuid, data={})

Constructs all the necessary attributes for the File object.

Parameters:

• api (AsyncGeoboxClient) – The AsyncGeoboxClient instance.

• uuid (str) – The UUID of the file.

• data (Dict, optional) – The data of the file.

__repr__()

Return a string representation of the File object.

Returns:

A string representation of the File object.

Return type:

str

property layers: List[Dict]

Get the layers of the file.

Returns:

The layers of the file.

Return type:

List[Dict]

Example

>>> from geobox import GeoboxClient
>>> client = GeoboxClient()
>>> file = File.get_file(client, uuid="12345678-1234-5678-1234-567812345678")
>>> file.layers

property file_type: FileType

Get the file type

Returns:

the file type enumeration

Return type:

FileType

Example

>>> from geobox.aio import AsyncGeoboxClient
>>> async with AsyncGeoboxClient() as client:
>>>     file = await File.get_file(client, uuid="12345678-1234-5678-1234-567812345678")
>>>     file.file_type

async classmethod upload_file(api, path, user_id=None, scan_archive=True)

[async] Upload a file to the GeoBox API.

Parameters:

• api (AsyncGeoboxClient) – The AsyncGeoboxClient instance.

• path (str) – The path to the file to upload.

• user_id (int, optional) – specific user. privileges required.

• scan_archive (bool, optional) – Whether to scan the archive for layers. default: True

Returns:

The uploaded file instance.

Return type:

AsyncFile

Raises:

• ValueError – If the file type is invalid.

• FileNotFoundError – If the file does not exist.

Example

>>> from geobox.aio import AsyncGeoboxClient
>>> from geobox.aio.file import AsyncFile
>>> async with AsyncGeoboxClient() as client:
>>>     file = await AsyncFile.upload_file(client, path='path/to/file.shp')
or
>>>     file = await client.upload_file(path='path/to/file.shp')

async classmethod get_files(api, **kwargs)

[async] Retrieves a list of files.

Parameters:

api (AsyncGeoboxClient) – The AsyncGeoboxClient instance for making requests.

Keyword Arguments:

• q (str) – query filter based on OGC CQL standard. e.g. “field1 LIKE ‘%GIS%’ AND created_at > ‘2021-01-01’”

• search (str) – search term for keyword-based searching among search_fields or all textual fields if search_fields does not have value. NOTE: if q param is defined this param will be ignored.

• search_fields (str) – comma separated list of fields for searching

• order_by (str) – comma separated list of fields for sorting results [field1 A|D, field2 A|D, …]. e.g. name A, type D.NOTE: “A” denotes ascending order and “D” denotes descending order.

• return_count (bool) – if true, the total number of results will be returned. default is False.

• skip (int) – number of results to skip. default is 0.

• limit (int) – number of results to return. default is 10.

• user_id (int) – filter by user id.

• shared (bool) – Whether to return shared files. default is False.

Returns:

A list of File objects or the total number of results.

Return type:

List[AsyncFile] | int

Example

>>> from geobox.aio import AsyncGeoboxClient
>>> from geobox.aio.file import AsyncFile
>>> async with AsyncGeoboxClient() as client:
>>>     files = await AsyncFile.get_files(client, search_fields='name', search='GIS', order_by='name', skip=10, limit=10)
or
>>>     files = await client.get_files(search_fields='name', search='GIS', order_by='name', skip=10, limit=10)

async classmethod get_file(api, uuid, user_id=None)

[async] Retrieves a file by its UUID.

Parameters:

• api (AsyncGeoboxClient) – The AsyncGeoboxClient instance.

• uuid (str) – The UUID of the file.

• user_id (int, optional) – specific user. privileges required.

Returns:

The retrieved file instance.

Return type:

AsyncFile

Raises:

NotFoundError – If the file with the specified UUID is not found.

Example

>>> from geobox.aio import AsyncGeoboxClient
>>> from geobox.aio.file import AsyncFile
>>> async with AsyncGeoboxClient() as client:
>>>     file = await AsyncFile.get_file(client, uuid="12345678-1234-5678-1234-567812345678")
or
>>>     file = await client.get_file(uuid="12345678-1234-5678-1234-567812345678")

async classmethod get_files_by_name(api, name, user_id=None)

[async] Get files by name

Parameters:

• api (AsyncGeoboxClient) – The AsyncGeoboxClient instance for making requests.

• name (str) – the name of the file to get

• user_id (int, optional) – specific user. privileges required.

Returns:

returns files that matches the given name

Return type:

List[AsyncFile]

Example

>>> from geobox.aio import AsyncGeoboxClient
>>> from geobox.aio.file import AsyncFile
>>> async with AsyncGeoboxClient() as client:
>>>     files = await AsyncFile.get_files_by_name(client, name='test')
or
>>>     files = await client.get_files_by_name(name='test')

_get_file_name(response)

Get the file name from the response.

Parameters:

response (aiohttp.ClientResponse) – The response of the request.

Returns:

The file name

Return type:

str

_create_progress_bar()

Creates a progress bar for the task.

Return type:

tqdm

async download(save_path=None, progress_bar=True, file_name=None, overwrite=False)

[async] Download a file and save it to the specified path.

Parameters:

• save_path (str, optional) – Path where the file should be saved. If not provided, it saves to the current working directory using the original filename and appropriate extension.

• progress_bar (bool, optional) – Whether to show a progress bar. default: True

• file_name (str, optional) – the downloaded file name.

• overwrite (bool, optional) – whether to overwrite the downloaded file if it exists on the save path. default is False.

Returns:

Path where the file was saved

Return type:

str

Raises:

• ValueError – If uuid is not set

• OSError – If there are issues with file operations

Example

>>> from geobox.aio import AsyncGeoboxClient
>>> from geobox.aio.file import AsyncFile
>>> async with AsyncGeoboxClient() as client:
>>>     file = await AsyncFile.get_file(client, uuid="12345678-1234-5678-1234-567812345678")
>>>     await file.download(save_path='path/to/save/')

async delete()

[async] Deletes the file.

Returns:

None

Return type:

None

Example

>>> from geobox.aio import AsyncGeoboxClient
>>> from geobox.aio.file import AsyncFile
>>> async with AsyncGeoboxClient() as client:
>>>     file = await AsyncFile.get_file(client, uuid="12345678-1234-5678-1234-567812345678")
>>>     await file.delete()

async publish(name, publish_as=None, input_geom_type=None, input_layer=None, input_dataset=None, user_id=None, input_srid=3857, file_encoding='UTF-8', replace_domain_codes_by_values=False, report_errors=True, as_terrain=False)

[async] Publishes a file as a layer.

Parameters:

• name (str) – The name of the layer.

• publish_as (PublishFileType, optional) – The type of layer to publish as.

• input_geom_type (InputGeomType, optional) – The geometry type of the layer.

• input_layer (str, optional) – The name of the input layer.

• input_dataset (str, optional) – The name of the input dataset.

• user_id (int, optional) – Specific user. privileges required.

• input_srid (int, optional) – The SRID of the layer. default is: 3857

• file_encoding (str, optional) – The encoding of the file. default is “utf-8”.

• replace_domain_codes_by_values (bool, optional) – Whether to replace domain codes by values. default is False.

• report_errors (bool, optional) – Whether to report errors. default is True.

• as_terrain (bool, optional) – Whether to publish as terrain. default is False.

Returns:

The task object.

Return type:

Task

Raises:

• ValueError – If the publish_as is not a valid PublishFileType.

• ValidationError – if the zipped file doesn’t have any layers to publish.

Example

>>> from geobox.aio import AsyncGeoboxClient
>>> from geobox.aio.file import AsyncFile
>>> async with AsyncGeoboxClient() as client:
>>>     file = await AsyncFile.get_file(client, uuid="12345678-1234-5678-1234-567812345678")
>>>     await file.publish(publish_as=PublishFileType.VECTOR,
...                         layer_name='my_layer',
...                         input_geom_type=InputGeomType.POINT,
...                         input_layer='layer1',
...                         input_dataset='dataset1',
...                         input_srid=4326,
...                         file_encoding='UTF-8')

async share(users)

[async] Shares the file with specified users.

Parameters:

users (List[User]) – The list of users objects to share the file with.

Returns:

None

Return type:

None

Example

>>> from geobox.aio import AsyncGeoboxClient
>>> from geobox.aio.file import AsyncFile
>>> async with AsyncGeoboxClient() as client:
>>>     file = await AsyncFile.get_file(client, uuid="12345678-1234-5678-1234-567812345678")
>>>     users = await client.search_users(search='John')
>>>     await file.share(users=users)

async unshare(users)

[async] Unshares the file with specified users.

Parameters:

users (List[AsyncUser]) – The list of users objects to unshare the file with.

Returns:

None

Return type:

None

Example

>>> from geobox.aio import AsyncGeoboxClient
>>> from geobox.aio.file import AsyncFile
>>> async with AsyncGeoboxClient() as client:
>>>     file = await AsyncFile.get_file(client, uuid="12345678-1234-5678-1234-567812345678")
>>>     await users = client.search_users(search='John')
>>>     await file.unshare(users=users)

async get_shared_users(search=None, skip=0, limit=10)

[async] Retrieves the list of users the file is shared with.

Parameters:

• search (str, optional) – The search query.

• skip (int, optional) – The number of users to skip.

• limit (int, optional) – The maximum number of users to retrieve.

Returns:

The list of shared users.

Return type:

List[AsyncUser]

Example

>>> from geobox.aio import AsyncGeoboxClient
>>> from geobox.aio.file import AsyncFile
>>> async with AsyncGeoboxClient() as client:
>>>     file = await AsyncFile.get_file(client, uuid="12345678-1234-5678-1234-567812345678")
>>>     await file.get_shared_users(search='John', skip=0, limit=10)