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
if TYPE_CHECKING:
from . import GeoboxClient
from .task import Task
[docs]
@dataclass
class DBCredentials:
"""Database connection credentials"""
db_type: DBType
"""Type of database (e.g., postgis, postgres, mysql)"""
host: str
"""e.g. localhost or 192.168.1.10"""
port: str
"""e.g. 5432"""
db_name: str
"""e.g. mygeodb"""
username: str
"""e.g. postgres"""
password: str
"""password"""
schemas: Optional[str] = None
"""
Comma-separated list of PostgreSQL schemas to include. Leave empty to scan all schemas
e.g. public, myapp (leave empty for all schemas)
"""
[docs]
def to_dict(self) -> Dict:
"""Convert to dict with enum values serialized"""
data = asdict(self)
data['db_type'] = self.db_type.value if type(self.db_type) == DBType else self.db_type
return data
[docs]
def __post_init__(self):
"""Validate credentials after initialization"""
if ((type(self.db_type) == str and self.db_type != 'postgis') or \
(type(self.db_type) == DBType and self.db_type.value != 'postgis')) and \
self.schemas is not None:
raise ValueError(f"schemas should only be provided for postgis, not {self.db_type}")
[docs]
class DatabaseTable(Base):
BASE_ENDPOINT = 'dbImport/'
[docs]
def __init__(
self,
api: 'GeoboxClient',
database: 'Database',
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"DatabaseTable(name={self.name}, geometry_type={self.geometry_type}, feature_count={self.feature_count})"
[docs]
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,
) -> 'Task':
"""
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:
Task: the import task object
Example:
>>> from geobox import GeoboxClient
>>> from geobox.db_connection import Database, DBCredentials, DBType
>>> client = GeoboxClient()
>>> creds = DBCredentials(...)
>>> tables = Database.get_database_tables(client, creds)
or
>>> tables = client.get_database_tables(creds)
>>> 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 = self.api.post(
endpoint=endpoint,
payload=data,
is_json=False,
)
return self.api.get_task(response['task_id'])
[docs]
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,
) -> 'Task':
"""
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:
Task: the import task object
Example:
>>> from geobox import GeoboxClient
>>> from geobox.db_connection import Database, DBCredentials, DBType
>>> client = GeoboxClient()
>>> creds = DBCredentials(...)
>>> tables = Database.get_database_tables(client, creds)
or
>>> tables = client.get_database_tables(creds)
>>> 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 = self.api.post(
endpoint=endpoint,
payload=data,
is_json=False,
)
return self.api.get_task(response['task_id'])
[docs]
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,
) -> 'Task':
"""
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:
Task: the import task object
Example:
>>> from geobox import GeoboxClient
>>> from geobox.db_connection import Database, DBCredentials, DBType
>>> client = GeoboxClient()
>>> creds = DBCredentials(...)
>>> layer = client.get_vectors()[0]
>>> tables = Database.get_database_tables(client, creds)
or
>>> tables = client.get_database_tables(creds)
>>> 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 = self.api.post(
endpoint=endpoint,
payload=data,
is_json=False,
)
return self.api.get_task(response['task_id'])
[docs]
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,
) -> 'Task':
"""
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:
Task: the import task object
Example:
>>> from geobox import GeoboxClient
>>> from geobox.db_connection import Database, DBCredentials, DBType
>>> client = GeoboxClient()
>>> creds = DBCredentials(...)
>>> layer = client.get_vectors()[0]
>>> tables = Database.get_database_tables(client, creds)
or
>>> tables = client.get_database_tables(creds)
>>> 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 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 self.import_non_spatial(
table_name=table_name,
out_table_name=out_name,
report_errors=report_errors,
user_id=user_id,
)
[docs]
class Database(Base):
BASE_ENDPOINT = 'dbImport/'
[docs]
def __init__(self, api: 'GeoboxClient', creds: 'DBCredentials'):
"""
Constructs all the necessary attributes for the Database object.
Args:
api (GeoboxClient): The GeoboxClient 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 Database object.
Returns:
str: A string representation of the Database object.
"""
return f"Database(db_type={self.creds.db_type}, db_name={self.creds.db_name}, user={self.creds.username})"
[docs]
@classmethod
def get_database_tables(
cls,
api: 'GeoboxClient',
creds: 'DBCredentials',
) -> List['DatabaseTable']:
"""
Get the list of tha database tables
Args:
api (GeoboxClient): The GeoboxClient instance.
creds (DBCredentials): the database connection credentials
Returns:
List[DatabaseTable]: list of the database tables
Example:
>>> from geobox import GeoboxClient
>>> from geobox.db_connection import Database, DBCredentials, DBType
>>> client = GeoboxClient()
>>> creds = DBCredentials(...)
>>> tables = Database.get_database_tables(client, creds)
or
>>> tables = client.get_database_tables(creds)
"""
db = Database(api=api, creds=creds)
endpoint = f"{cls.BASE_ENDPOINT}test-connection/"
response = api.post(
endpoint=endpoint,
payload=clean_data(creds.to_dict()),
is_json=False,
)
layers = [DatabaseTable(api=api, data=layer, database=db) for layer in response['layers']] if response.get('layers') else []
return layers