from dataclasses import dataclass, asdict
from typing import Dict, List, Optional, TYPE_CHECKING
from .base import Base
from ..enums import DBType
from ..utils import clean_data
from ..db_connection import DBCredentials
if TYPE_CHECKING:
from . import AsyncGeoboxClient
from .task import AsyncTask
[docs]
class AsyncDatabaseTable(Base):
BASE_ENDPOINT = 'dbImport/'
[docs]
def __init__(
self,
api: 'AsyncGeoboxClient',
database: 'AsyncDatabase',
data: Optional[Dict] = None,
):
"""
Constructs all the necessary attributes for the DatabaseTable object.
Args:
api (GeoboxClient): The GeoboxClient instance.
database (Database): the Database instance.
data (Dict, optional): The data of the table.
"""
super().__init__(api=api, data=data)
self.database = database
[docs]
def __repr__(self) -> str:
"""
Return a string representation of the DatabaseTable object.
Returns:
str: A string representation of the DatabaseTable object.
"""
return f"AsyncDatabaseTable(name={self.name}, geometry_type={self.geometry_type}, feature_count={self.feature_count})"
[docs]
async def import_spatial(
self,
table_name: Optional[str] = None,
out_layer_name: Optional[str] = None,
input_srid: Optional[int] = None,
input_geom_type: Optional[str] = None,
report_errors: bool = False,
user_id: Optional[int] = None,
) -> 'AsyncTask':
"""
[async] Import a spatial table/layer from an external database and publish it as a vector layer.
Args:
table_name (str, optional): Source table or layer name in the external database
out_layer_name (str, optional): Name for the new vector layer to create
input_srid (int, optional): Source coordinate reference system EPSG code
input_geom_type (str, optional): Force a specific geometry type
report_errors (bool, optional): Include per-feature import errors in the result. default: False
user_id (int, optional): specific user. privileges required.
Returns:
AsyncTask: the import task object
Example:
>>> from geobox.aio import AsyncGeoboxClient
>>> from geobox.aio.db_connection import AsyncDatabase, DBCredentials, DBType
>>> async with AsyncGeoboxClient() as client:
>>> creds = DBCredentials(...)
>>> tables = await AsyncDatabase.get_database_tables(client, creds)
or
>>> tables = await client.get_database_tables(creds)
>>> await tables[0].import_spatial()
"""
endpoint = f"{self.BASE_ENDPOINT}import/"
creds = self.database.creds.to_dict()
data = clean_data({
"table_name": table_name if table_name else self.name,
"layer_name": out_layer_name if out_layer_name else self.name,
"input_srid": input_srid,
"input_geom_type": input_geom_type,
"report_errors": report_errors,
"user_id": user_id,
**creds,
})
response = await self.api.post(
endpoint=endpoint,
payload=data,
is_json=False,
)
return await self.api.get_task(response['task_id'])
[docs]
async def import_non_spatial(
self,
table_name: Optional[str] = None,
out_table_name: Optional[str] = None,
report_errors: bool = False,
user_id: Optional[int] = False,
) -> 'AsyncTask':
"""
[async] Import a non-spatial table from an external database and publish it as a Geobox Table.
Args:
table_name (str, optional): Source table name in the external database
out_table_name (str, optional): Name for the new table to create
report_errors (bool, optional): Include per-row import errors in the result. default: False
user_id (int, optional): specific user. privileges required.
Returns:
AsyncTask: the import task object
Example:
>>> from geobox.aio import AsyncGeoboxClient
>>> from geobox.aio.db_connection import AsyncDatabase, DBCredentials, DBType
>>> async with AsyncGeoboxClient() as client:
>>> creds = DBCredentials(...)
>>> tables = await AsyncDatabase.get_database_tables(client, creds)
or
>>> tables = await client.get_database_tables(creds)
>>> await tables[0].import_non_spatial()
"""
endpoint = f"{self.BASE_ENDPOINT}import-table/"
creds = self.database.creds.to_dict()
data = clean_data({
"table_name": table_name if table_name else self.name,
"out_table_name": out_table_name if out_table_name else self.name,
"report_errors": report_errors,
"user_id": user_id,
**creds,
})
response = await self.api.post(
endpoint=endpoint,
payload=data,
is_json=False,
)
return await self.api.get_task(response['task_id'])
[docs]
async def import_spatial_into_layer(
self,
layer_uuid: str,
table_name: Optional[str] = None,
is_view: bool = False,
input_srid: Optional[int] = None,
input_geom_type: Optional[str] = None,
report_errors: bool = False,
user_id: Optional[int] = None,
) -> 'AsyncTask':
"""
[async] Import a spatial table/layer from an external database and append its features into an existing vector layer identified by layer_uuid.
Args:
layer_uuid (str): UUID of the existing vector layer to import into
table_name (str, optional): Source table or layer name in the external database
is_view (bool, optional): Whether the target layer is a vector layer view. default: False
input_srid (int, optional): Source CRS EPSG code
input_geom_type (str, optional): Force a specific geometry type
report_errors (bool, optional): Include per-feature import errors in the result. default: False
user_id (int, optional): specific user. privileges required.
Returns:
AsyncTask: the import task object
Example:
>>> from geobox.aio import AsyncGeoboxClient
>>> from geobox.aio.db_connection import AsyncDatabase, DBCredentials, DBType
>>> async with AsyncGeoboxClient() as client:
>>> creds = DBCredentials(...)
>>> layer = await client.get_vectors()[0]
>>> tables = await AsyncDatabase.get_database_tables(client, creds)
or
>>> tables = await client.get_database_tables(creds)
>>> await tables[0].import_spatial_into_layer(layer_uuid=layer.uuid)
"""
endpoint = f"{self.BASE_ENDPOINT}import-into-layer/"
creds = self.database.creds.to_dict()
data = clean_data({
"table_name": table_name if table_name else self.name,
"layer_uuid": layer_uuid,
"is_view": is_view,
"input_srid": input_srid,
"input_geom_type": input_geom_type,
"report_errors": report_errors,
"user_id": user_id,
**creds,
})
response = await self.api.post(
endpoint=endpoint,
payload=data,
is_json=False,
)
return await self.api.get_task(response['task_id'])
[docs]
async def import_table(
self,
table_name: Optional[str] = None,
out_name: Optional[str] = None,
input_srid: Optional[int] = None,
input_geom_type: Optional[str] = None,
report_errors: bool = False,
user_id: Optional[int] = None,
) -> 'AsyncTask':
"""
[async] Import a spatial table/layer from an external database and append its features into an existing vector layer identified by layer_uuid.
This method acts as a dispatcher that automatically routes the import request
to the appropriate method (import_spatial for tables with geometry columns or
import_non_spatial for tables without geometry).
Args:
table_name (str, optional): Source table or layer name in the external database.
out_name (str, optional): Name for the output resource (vector layer or table) in Geobox.
input_srid (int, optional): Source CRS EPSG code
input_geom_type (str, optional): Force a specific geometry type
report_errors (bool, optional): Include per-feature import errors in the result. default: False
user_id (int, optional): specific user. privileges required.
Returns:
AsyncTask: the import task object
Example:
>>> from geobox.aio import AsyncGeoboxClient
>>> from geobox.aio.db_connection import AsyncDatabase, DBCredentials, DBType
>>> async with AsyncGeoboxClient() as client:
>>> creds = DBCredentials(...)
>>> layer = await client.get_vectors()[0]
>>> tables = await AsyncDatabase.get_database_tables(client, creds)
or
>>> tables = await client.get_database_tables(creds)
>>> await tables[0].import()
See Also:
- import_spatial(): For explicitly importing spatial tables
- import_non_spatial(): For explicitly importing non-spatial tables
- import_spatial_into_layer(): For importing into an existing layer
"""
if self.has_geometry:
return await self.import_spatial(
table_name=table_name,
out_layer_name=out_name,
input_srid=input_srid,
input_geom_type=input_geom_type,
report_errors=report_errors,
user_id=user_id,
)
else:
return await self.import_non_spatial(
table_name=table_name,
out_table_name=out_name,
report_errors=report_errors,
user_id=user_id,
)
[docs]
class AsyncDatabase(Base):
BASE_ENDPOINT = 'dbImport/'
[docs]
def __init__(self, api: 'AsyncGeoboxClient', creds: 'DBCredentials'):
"""
Constructs all the necessary attributes for the Database object.
Args:
api (AsyncGeoboxClient): The AsyncGeoboxClient instance.
creds (DBCredentials): the database connection credentials
"""
super().__init__(api=api)
self.creds = creds
[docs]
def __repr__(self) -> str:
"""
Return a string representation of the AsyncDatabase object.
Returns:
str: A string representation of the AsyncDatabase object.
"""
return f"AsyncDatabase(db_type={self.creds.db_type}, db_name={self.creds.db_name}, user={self.creds.username})"
[docs]
@classmethod
async def get_database_tables(
cls,
api: 'AsyncGeoboxClient',
creds: 'DBCredentials',
) -> List['AsyncDatabaseTable']:
"""
[async] Get the list of tha database tables
Args:
api (AsyncGeoboxClient): The AsyncGeoboxClient instance.
creds (DBCredentials): the database connection credentials
Returns:
List[AsyncDatabaseTable]: list of the database tables
Example:
>>> from geobox.aio import AsyncGeoboxClient
>>> from geobox.aio.db_connection import AsyncDatabase, DBCredentials, DBType
>>> async with AsyncGeoboxClient() as client:
>>> creds = DBCredentials(...)
>>> tables = await AsyncDatabase.get_database_tables(client, creds)
or
>>> tables = await client.get_database_tables(creds)
"""
db = AsyncDatabase(api=api, creds=creds)
endpoint = f"{cls.BASE_ENDPOINT}test-connection/"
response = await api.post(
endpoint=endpoint,
payload=clean_data(creds.to_dict()),
is_json=False,
)
layers = [AsyncDatabaseTable(api=api, data=layer, database=db) for layer in response['layers']] if response.get('layers') else []
return layers