from typing import List, Dict, Optional, TYPE_CHECKING, Union
from dataclasses import dataclass
from geobox.exception import NotFoundError, ValidationError
from geobox.utils import clean_data
from .base import AsyncBase
from ..enums import RelationshipCardinality
from ..exception import NotFoundError, ValidationError
if TYPE_CHECKING:
from . import AsyncGeoboxClient
from .table import AsyncTable, AsyncTableField
from .vectorlayer import AsyncVectorLayer
from .field import AsyncField
[docs]
@dataclass(frozen=True)
class RelationshipEndpoint:
"""
Represents one endpoint (source or target) of a relationship
Args:
table (AsyncTable | AsyncVectorLayer): The source or target table or vector layer
field (AsyncTableField | AsyncField | str): The field name or object on the source/target entity
fk_field (AsyncTableField | AsyncField | str, optional): The foreign key field name or object on the relation table. (Required for Many-to-Many relationships)
"""
table: Union['AsyncTable', 'AsyncVectorLayer']
field: Union['AsyncTableField', 'AsyncField', str]
fk_field: Optional[Union['AsyncTableField', 'AsyncField', str]] = None
[docs]
class AsyncRelationship(AsyncBase):
BASE_ENDPOINT = 'relationships/'
[docs]
def __init__(self,
api: 'AsyncGeoboxClient',
uuid: str,
data: Optional[Dict] = {},
):
"""
Initialize a relationship instance.
Args:
api (AsyncGeoboxClient): The GeoboxClient instance for making requests.
uuid (str): The unique identifier for the relationship.
data (Dict): The response data of the table.
"""
super().__init__(api=api, uuid=uuid, data=data)
[docs]
def __repr__(self) -> str:
"""
Return a string representation of the AsyncRelationship.
Returns:
str: The string representation of the AsyncRelationship.
"""
return f"AsyncRelationship(uuid={self.uuid}, name={self.data.get('relation_name') or self.data.get('name')}, cardinality={self.data.get('relation_cardinality') or self.data.get('cardinality')})"
@property
def cardinality(self) -> RelationshipCardinality:
"""
Return:
RelationshipCardinality: the relationship cardinality
"""
return RelationshipCardinality(self.relation_cardinality)
[docs]
@classmethod
async def get_relationships(
cls,
api: 'AsyncGeoboxClient',
**kwargs,
) -> Union[List['AsyncRelationship'], int]:
"""
[async] Get a list of relationships with optional filtering and pagination.
Args:
api (AsyncGeoboxClient): The GeoboxClient instance for making requests.
Keyword Args:
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): Whether to return total count. default: False.
skip (int): Number of items to skip. default: 0
limit (int): Number of items to return. default: 10
user_id (int): Specific user. privileges required
shared (bool): Whether to return shared tables. default: False
Returns:
List[AsyncRelationship] | int: A list of relationship instances or the total number of relationships.
Example:
>>> from geobox.aio import AsyncGeoboxClient
>>> from geobox.aio.table import AsyncRelatinship
>>> async with AsyncGeoboxClient() as client:
>>> relationships = await client.get_relationships(q="name LIKE '%My relationship%'")
or
>>> relationships = await Table.get_relationships(client, q="name LIKE '%My relationship%'")
"""
params = {
'f': 'json',
'q': kwargs.get('q'),
'search': kwargs.get('search'),
'search_fields': kwargs.get('search_fields'),
'order_by': kwargs.get('order_by'),
'return_count': kwargs.get('return_count', False),
'skip': kwargs.get('skip', 0),
'limit': kwargs.get('limit', 10),
'user_id': kwargs.get('user_id'),
'shared': kwargs.get('shared', False),
}
return await super()._get_list(api, cls.BASE_ENDPOINT, params, factory_func=lambda api, item: AsyncRelationship(api, item['uuid'], item))
[docs]
@classmethod
async def create_relationship(
cls,
api: 'AsyncGeoboxClient',
name: str,
cardinality: RelationshipCardinality,
*,
source: 'RelationshipEndpoint',
target: 'RelationshipEndpoint',
relation_table: Optional['AsyncTable'] = None,
display_name: Optional[str] = None,
description: Optional[str] = None,
user_id: Optional[int] = None,
) -> 'AsyncRelationship':
"""
[async] Create a new AsyncRelationship
Args:
api (AsyncGeoboxClient): The GeoboxClient instance for making requests.
name (str): name of the relationship
cardinality (RelationshipCardinality): One to One, One to Many, or Many to Many
Keyword Args:
source (RelationshipEndpoint): Definition of the source side of the relationship, including the table (or layer), field, and foreign-key field
target (RelationshipEndpoint): Definition of the target side of the relationship, including the table (or layer), field, and foreign-key field
relation_table (AsyncTable, optional): The table that stores the relationship metadata or join records. (Required for Many-to-Many relationships)
display_name (str, optional): Human-readable name for the relationship
description (str, optional): the description of the relationship
user_id (int, optional): Specific user. privileges required.
Returns:
AsyncRelationship: a relationship instance
Example:
>>> from geobox.aio import AsyncGeoboxClient
>>> from geobox.aio.table import AsyncRelationship, RelationshipEndpoint, RelationshipCardinality
>>> async with AsyncGeoboxClient() as client:
>>> source = RelationshipEndpoint(
... table=client.get_table_by_name('owner'),
... field="name", # on source table
... fk_field="book_name", # on relation table
... )
>>> target = RelationshipEndpoint(
... table=client.get_table_by_name('parcel'),
... field="name", # on target table
... fk_field="author_name", # on relation table
... )
>>> relationship = await client.create_relationship(
... name="owner_parcel",
... cardinality=RelationshipCardinality.ManytoMany,
... source=source,
... target=target,
... relation_table=client.get_table_by_name('owner_parcel'),
... )
or
>>> relationship = await AsyncRelationship.create_relationship(
... client,
... name="owner_parcel",
... cardinality=RelationshipCardinality.ManytoMany,
... source=source,
... target=target,
... relation_table=client.get_table_by_name('owner_parcel'),
... )
"""
data = {
"relation_name": name,
"relation_display_name": display_name,
"relation_description": description,
"relation_cardinality": cardinality.value,
"relation_table_id": relation_table.id if relation_table else None,
"relation_table_search": f"{relation_table.__class__.__name__.split('Async')[-1]}/{relation_table.name}" if relation_table else None,
"source_field_name": source.field if type(source.field) == str else source.field.name,
"source_fk_name": (source.fk_field if not source.fk_field or type(source.fk_field) == str else source.fk_field.name) if relation_table else None,
"source_id": source.table.id,
"source_search": f"{source.table.__class__.__name__.split('Async')[-1]}/{source.table.name}",
"source_type": str(source.table.__class__.__name__.split('Async')[-1]),
"target_field_name": target.field if type(target.field) == str else target.field.name,
"target_fk_name": (target.fk_field if not target.fk_field or type(target.fk_field) == str else target.fk_field.name) if relation_table else None,
"target_id": target.table.id,
"target_search": f"{target.table.__class__.__name__.split('Async')[-1]}/{target.table.name}",
"target_type": str(target.table.__class__.__name__.split('Async')[-1]),
"user_id": user_id,
}
return await super()._create(api, cls.BASE_ENDPOINT, data, factory_func=lambda api, item: AsyncRelationship(api, item['uuid'], item))
[docs]
@classmethod
async def get_relationship(
cls,
api: 'AsyncGeoboxClient',
uuid: str,
user_id: Optional[int] = None,
) -> 'AsyncRelationship':
"""
[async] Get a relationship by UUID.
Args:
api (AsyncGeoboxClient): The AsyncGeoboxClient instance for making requests.
uuid (str): The UUID of the relationship to get.
user_id (int, optional): Specific user. privileges required.
Returns:
AsyncRelationship: The AsyncRelationship object.
Raises:
NotFoundError: If the AsyncRelationship with the specified UUID is not found.
Example:
>>> from geobox.aio import AsyncGeoboxClient
>>> from geobox.aio.table import AsyncRelationship
>>> async with AsyncGeoboxClient() as client:
>>> relationship = await client.get_relationship(uuid="12345678-1234-5678-1234-567812345678")
or
>>> relationship = await AsyncRelationship.get_relationship(client, uuid="12345678-1234-5678-1234-567812345678")
"""
params = {
'f': 'json',
'user_id': user_id,
}
return await super()._get_detail(api, cls.BASE_ENDPOINT, uuid, params, factory_func=lambda api, item: AsyncRelationship(api, item['uuid'], item))
[docs]
@classmethod
async def get_relationship_by_name(
cls,
api: 'AsyncGeoboxClient',
name: str,
user_id: Optional[int] = None,
) -> Union['AsyncRelationship', None]:
"""
[async] Get a relationship by name
Args:
api (AsyncGeoboxClient): The GeoboxClient instance for making requests.
name (str): the name of the relationship to get
user_id (int, optional): specific user. privileges required.
Returns:
AsyncRelationship | None: returns the relationship if a relationship matches the given name, else None
Example:
>>> from geobox.aio import AsyncGeoboxClient
>>> from geobox.aio.table import AsyncRelationship
>>> async with AsyncGeoboxClient() as client:
>>> relationship = await client.get_relationship_by_name(name='test')
or
>>> relationship = await AsyncRelationship.get_relationship_by_name(client, name='test')
"""
relationships = await cls.get_relationships(api, q=f"name = '{name}'", user_id=user_id)
if relationships and relationships[0].relation_name == name:
return relationships[0]
else:
return None
[docs]
async def update(self,
*,
name: Optional[str] = None,
cardinality: Optional[RelationshipCardinality] = None,
source: Optional['RelationshipEndpoint'] = None,
target: Optional['RelationshipEndpoint'] = None,
relation_table: Optional['AsyncTable'] = None,
display_name: Optional[str] = None,
description: Optional[str] = None,
) -> Dict:
"""
[async] Update the relationship.
Keyword Args:
name (str): The name of the relationship.
cardinality (RelationshipCardinality): One to One, One to Many, or Many to Many
source (RelationshipEndpoint): Definition of the source side of the relationship, including the table (or layer), field, and foreign-key field
target (RelationshipEndpoint): Definition of the target side of the relationship, including the table (or layer), field, and foreign-key field
relation_table (AsyncTable, optional): The table that stores the relationship metadata or join records. (Required for Many-to-Many relationships)
display_name (str, optional): Human-readable name for the relationship
description (str, optional): the description of the relationship
Returns:
Dict: The updated relationship data.
Raises:
ValidationError: If the relationship data is invalid.
Example:
>>> from geobox.aio import AsyncGeoboxClient
>>> from geobox.aio.table import AsyncRelationship
>>> async with AsyncGeoboxClient() as client:
>>> relationship = await AsyncRelationship.get_relationship(client, uuid="12345678-1234-5678-1234-567812345678")
>>> await relationship.update(display_name="New Display Name")
"""
# Validate relationships
if cardinality == RelationshipCardinality.ManytoMany and not relation_table:
raise ValidationError("relation_table is required for Many-to-Many relationships")
# Helper to extract field name
def get_field_name(field) -> Optional[str]:
if field is None:
return None
return field if isinstance(field, str) else field.name
# Helper to get table info
def get_table_info(table) -> Dict[str, str]:
if table is None:
return {}
return {
"id": table.id,
"search": f"{table.__class__.__name__}/{table.name}",
"type": table.__class__.__name__
}
# Build payload
data = {
"relation_name": name,
"relation_display_name": display_name,
"relation_description": description,
"relation_cardinality": cardinality.value if cardinality else None,
}
# Add relation table info if present
if relation_table:
data.update({
"relation_table_id": relation_table.id,
"relation_table_search": f"{relation_table.__class__.__name__}/{relation_table.name}"
})
# Add source info
if source:
source_table_info = get_table_info(source.table)
data.update({
"source_field_name": get_field_name(source.field),
"source_id": source_table_info.get("id"),
"source_search": source_table_info.get("search"),
"source_type": source_table_info.get("type"),
})
if relation_table:
data["source_fk_name"] = get_field_name(source.fk_field)
# Add target info
if target:
target_table_info = get_table_info(target.table)
data.update({
"target_field_name": get_field_name(target.field),
"target_id": target_table_info.get("id"),
"target_search": target_table_info.get("search"),
"target_type": target_table_info.get("type"),
})
if relation_table:
data["target_fk_name"] = get_field_name(target.fk_field)
return await super()._update(self.endpoint, data)
[docs]
async def delete(self) -> None:
"""
[async] Delete the AsyncRelationship
Returns:
None
Example:
>>> from geobox.aio import AsyncGeoboxClient
>>> from geobox.aio.table import AsyncRelationship
>>> async with AsyncGeoboxClient() as client:
>>> relationship = await AsyncRelationship.get_relationship(client, uuid="12345678-1234-5678-1234-567812345678")
>>> await relationship.delete()
"""
await super()._delete(self.endpoint)
[docs]
async def get_source(self) -> Union['AsyncTable', 'AsyncVectorLayer']:
"""
[async] Get the source table or layer
Returns:
AsyncTable | AsyncVectorLayer: the source table or layer
Raises:
NotFoundError: if the table or layer has been deleted
Example:
>>> from geobox.aio import AsyncGeoboxClient
>>> async with AsyncGeoboxClient() as client:
>>> relationship = await client.get_relationship(uuid="12345678-1234-5678-1234-567812345678")
>>> source_table = await relationship.get_source()
"""
try:
result = []
if self.source_type == 'Table':
result = await self.api.get_tables(
q=f"id = {self.source_id}"
)
elif self.source_type == 'VectorLayer':
result = await self.api.get_vectors(
q=f"id = {self.source_id}"
)
source = next(source for source in result if source.id == self.source_id)
return source
except StopIteration:
raise NotFoundError("Table not found!")
[docs]
async def get_target(self) -> Union['AsyncTable', 'AsyncVectorLayer']:
"""
[async] Get the target table or layer
Returns:
AsyncTable: the target table or layer
Raises:
NotFoundError: if the table or layer has been deleted
Example:
>>> from geobox.aio import AsyncGeoboxClient
>>> async with AsyncGeoboxClient() as client:
>>> relationship = await client.get_relationship(uuid="12345678-1234-5678-1234-567812345678")
>>> target_table = await relationship.get_target()
"""
try:
result = []
if self.target_type == 'Table':
result = await self.api.get_tables(
q=f"id = {self.target_id}"
)
elif self.target_type == 'VectorLayer':
result = await self.api.get_vectors(
q=f"id = {self.target_id}"
)
target = next(target for target in result if target.id == self.target_id)
return target
except StopIteration:
raise NotFoundError("Table not found!")
[docs]
async def get_relation_table(self) -> 'AsyncTable':
"""
[async] Get the relation table
Returns:
AsyncTable: the relation table
Raises:
ValueError: If the relationship is not Many-to-Many and thus has no relation table
NotFoundError: if the table has been deleted
Example:
>>> from geobox.aio import AsyncGeoboxClient
>>> async with AsyncGeoboxClient() as client:
>>> relationship = await client.get_relationship(uuid="12345678-1234-5678-1234-567812345678")
>>> relation_table = await relationship.get_relation_table()
"""
try:
if not self.relation_table_id:
raise ValueError(
"Relationship is not Many-to-Many. "
"Relation tables are only used for Many-to-Many cardinality."
)
result = await self.api.get_tables(
q=f"id = {self.relation_table_id}"
)
table = next(table for table in result if table.id == self.relation_table_id)
return table
except StopIteration:
raise NotFoundError("Table not found!")
[docs]
async def associate_records(
self,
source_id: int,
*,
target_ids: Optional[List[int]] = None,
q: Optional[str] = None,
) -> Dict:
"""
[async] Create relationships between the source record and target records
Args:
source_id (int): the id of feature/row in the source layer/table
target_ids (List[int], optional): a list of target record ids to be associated with the current record
q (str, optional): query filter on target layer or table to select which target features or rows that are going to be related to the current record
Returns:
Dict: the record association result
Example:
>>> from geobox.aio import AsyncGeoboxClient
>>> async with AsyncGeoboxClient() as client:
>>> relationship = await client.get_relationship(uuid="12345678-1234-5678-1234-567812345678")
>>> result = await relationship.associate_records(
... source_id=1,
... target_ids=[1, 2, 3],
... q="name LIKE '%_school'",
... )
"""
data = {
'source_id': source_id,
'target_ids': ', '.join([str(i) for i in target_ids]),
'q': q,
}
endpoint = f"{self.endpoint}associateRecords/"
return await self.api.post(
endpoint,
clean_data(data),
is_json=False,
)
[docs]
async def disassociate_records(
self,
source_id: int,
*,
target_ids: Optional[List[int]] = None,
q: Optional[str] = None,
) -> Dict:
"""
[async] Remove relationships between the source record and target records
Args:
source_id (int): the id of feature/row in the source layer/table
target_ids (List[int], optional): a list of target record ids to be disassociated with the current record
q (str, optional): query filter on target layer or table to select which target features or rows that are going to be related to the current record
Returns:
Dict: the record disassociation result
Example:
>>> from geobox.aio import AsyncGeoboxClient
>>> async with AsyncGeoboxClient() as client:
>>> relationship = await client.get_relationship(uuid="12345678-1234-5678-1234-567812345678")
>>> result = await relationship.disassociate_records(
... source_id=1,
... target_ids=[1, 2, 3],
... q="name LIKE '%_school'",
... )
"""
data = {
'source_id': source_id,
'target_ids': ', '.join([str(i) for i in target_ids]),
'q': q,
}
endpoint = f"{self.endpoint}disassociateRecords/"
return await self.api.post(
endpoint,
clean_data(data),
is_json=False,
)