from urllib.parse import urljoin
from typing import Optional, List, Dict, Any, TYPE_CHECKING, Union
from .base import AsyncBase
from ..enums import FeatureType
if TYPE_CHECKING:
from .vectorlayer import VectorLayer, AsyncVectorLayer
from .table import AsyncTable, AsyncTableRow, AsyncRelationship
[docs]
class AsyncFeature(AsyncBase):
BASE_SRID = 3857
[docs]
def __init__(self,
layer: 'VectorLayer',
srid: Optional[int] = 3857,
data: Optional[Dict] = {}):
"""
Constructs all the necessary attributes for the Feature object.
Args:
layer (VectorLayer): The vector layer this feature belongs to
srid (int, optional): The Spatial Reference System Identifier (default is 3857)
data (Dict, optional): The feature data contains the feature geometry and properties
"""
super().__init__(api=layer.api)
self.layer = layer
self._srid = srid
self.data = data or {
"type": "Feature",
"geometry": {},
"properties": {}
}
self.original_geometry = self.data.get('geometry')
self.endpoint = urljoin(layer.endpoint, f'features/{self.data.get("id")}/') if self.data.get('id') else None
[docs]
def __dir__(self) -> List[str]:
"""
Return a list of available attributes for the Feature object.
This method extends the default dir() behavior to include:
- All keys from the feature data dictionary
- All keys from the geometry dictionary
- All keys from the properties dictionary
This allows for better IDE autocompletion and introspection of feature attributes.
Returns:
list: A list of attribute names available on this Feature object.
"""
return super().__dir__() + list(self.data.keys()) + list(self.data.get('geometry').keys()) + list(self.data.get('properties').keys())
[docs]
def __repr__(self) -> str:
"""
Return a string representation of the Feature object.
Returns:
str: A string representation of the Feature object.
"""
feature_id = getattr(self, "id", "-1")
return f"AsyncFeature(id={feature_id}, type={self.feature_type})"
[docs]
def __getattr__(self, name: str) -> Any:
"""
Get an attribute from the resource.
Args:
name (str): The name of the attribute
"""
if name in self.data:
return self.data.get(name)
# elif name in self.data['geometry']:
# return self.data['geometry'].get(name)
elif name in self.data['properties']:
return self.data['properties'].get(name)
raise AttributeError(f"Feature has no attribute {name}")
@property
def srid(self) -> int:
"""
Get the Spatial Reference System Identifier (SRID) of the feature.
Returns:
int: The SRID of the feature.
Example:
>>> from geobox.aio import AsyncGeoboxClient
>>> async with AsyncGeoboxClient() as client:
>>> layer = await client.get_vector(uuid="12345678-1234-5678-1234-567812345678")
>>> feature = await layer.get_feature(id=1)
>>> feature.srid # 3857
"""
return self._srid
@property
def feature_type(self) -> 'FeatureType':
"""
Get the type of the feature.
Returns:
FeatureType: The type of the feature.
Example:
>>> from geobox.aio import AsyncGeoboxClient
>>> async with AsyncGeoboxClient() as client:
>>> layer = await client.get_vector(uuid="12345678-1234-5678-1234-567812345678")
>>> feature = await layer.get_feature(id=1)
>>> feature.feature_type
"""
return FeatureType(self.data.get('geometry').get('type')) if self.data.get('geometry') else None
@property
def coordinates(self) -> List[float]:
"""
Get the coordinates of the ferepoature.
Returns:
list: The coordinates of the feature.
Example:
>>> from geobox.aio import AsyncGeoboxClient
>>> async with AsyncGeoboxClient() as client:
>>> layer = await client.get_vector(uuid="12345678-1234-5678-1234-567812345678")
>>> feature = await layer.get_feature(id=1)
>>> feature.coordinates
"""
return self.data.get('geometry').get('coordinates') if self.data.get('geometry') else None
@coordinates.setter
def coordinates(self, value: List[float]) -> None:
"""
Set the coordinates of the feature.
Args:
value (list): The coordinates to set.
Example:
>>> from geobox.aio import AsyncGeoboxClient
>>> async with AsyncGeoboxClient() as client:
>>> layer = await client.get_vector(uuid="12345678-1234-5678-1234-567812345678")
>>> feature = await layer.get_feature(id=1)
>>> feature.coordinates = [10, 20]
"""
self.data['geometry']['coordinates'] = value
@property
async def length(self) -> float:
"""
[async] Returns the length of thefeature geometry (geometry package extra is required!)
Returns:
float: the length of thefeature geometry
Example:
>>> from geobox.aio import AsyncGeoboxClient
>>> async with AsyncGeoboxClient() as client:
>>> layer = await client.get_vector(uuid="12345678-1234-5678-1234-567812345678")
>>> feature = await layer.get_feature(id=1)
>>> await feature.length
"""
try:
return self.geom_length
except AttributeError:
endpoint = f'{self.endpoint}length'
return await self.api.get(endpoint)
@property
async def area(self) -> float:
"""
[async] Returns the area of thefeature geometry (geometry package extra is required!)
Returns:
float: the area of thefeature geometry
Example:
>>> from geobox.aio import AsyncGeoboxClient
>>> async with AsyncGeoboxClient() as client:
>>> layer = await client.get_vector(uuid="12345678-1234-5678-1234-567812345678")
>>> feature = await layer.get_feature(id=1)
>>> await feature.area
"""
try:
return self.geom_area
except AttributeError:
endpoint = f'{self.endpoint}area'
return await self.api.get(endpoint)
[docs]
async def save(self) -> None:
"""
[async] Save the feature. Creates a new feature if feature_id is None, updates existing feature otherwise.
Returns:
None
Example:
>>> from geobox.aio import AsyncGeoboxClient
>>> async with AsyncGeoboxClient() as client:
>>> layer = await client.get_vector(uuid="12345678-1234-5678-1234-567812345678")
>>> feature = await layer.get_feature(id=1)
>>> feature.properties['name'] = 'New Name'
>>> await feature.save()
"""
data = self.data.copy()
srid = self.srid
try:
if self.id:
await self.update(self.data, srid=srid if srid != self.BASE_SRID else None)
except AttributeError:
endpoint = urljoin(self.layer.endpoint, 'features/')
if self.srid != self.BASE_SRID:
endpoint = f"{endpoint}?in_srid={self.srid}"
request_data = self.data.copy()
response = await self.layer.api.post(endpoint, request_data)
self.endpoint = urljoin(self.layer.endpoint, f'features/{response["id"]}/')
self.data.update(response)
self.data['geometry'] = data['geometry']
self._srid = srid
[docs]
async def delete(self) -> None:
"""
[async] Delete the feature.
Returns:
None
Example:
>>> from geobox.aio import AsyncGeoboxClient
>>> async with AsyncGeoboxClient() as client:
>>> layer = await client.get_vector(uuid="12345678-1234-5678-1234-567812345678")
>>> feature = await layer.get_feature(id=1)
>>> await feature.delete()
"""
await super()._delete(self.endpoint)
[docs]
async def update(
self,
geojson: Dict,
srid: Optional[int] = None,
) -> Dict:
"""
[async] Update the feature data property.
Args:
geojson (Dict): The GeoJSON data for the feature
srid (int, optional): the input geometry srid
Returns:
Dict: The response from the API.
Example:
>>> from geobox.aio import AsyncGeoboxClient
>>> async with AsyncGeoboxClient() as client:
>>> layer = await client.get_vector(uuid="12345678-1234-5678-1234-567812345678")
>>> feature = await layer.get_feature(id=1)
>>> geojson = {
... "geometry": {
... "type": "Point",
... "coordinates": [10, 20]
... }
... }
>>> await feature.update(geojson)
"""
endpoint = self.endpoint
if srid is not None:
endpoint = f"{endpoint}?in_srid={srid}"
elif self.srid != self.BASE_SRID:
endpoint = f"{endpoint}?in_srid={self.srid}"
await super()._update(endpoint, geojson, clean=False)
self.data['geometry'] = geojson['geometry']
self._srid = self.srid
return self.data
[docs]
@classmethod
async def create_feature(cls, layer: 'VectorLayer', geojson: Dict, srid: int = 3857) -> 'AsyncFeature':
"""
[async] Create a new feature in the vector layer.
Args:
layer (VectorLayer): The vector layer to create the feature in
geojson (Dict): The GeoJSON data for the feature
srid (int, optional): the feature srid. default: 3857
Returns:
AsyncFeature: The created feature instance
Example:
>>> from geobox.aio import AsyncGeoboxClient
>>> async with AsyncGeoboxClient() as client:
>>> layer = await client.get_vector(uuid="12345678-1234-5678-1234-567812345678")
>>> geojson = {
... "type": "Feature",
... "geometry": {"type": "Point", "coordinates": [10, 20]},
... "properties": {"name": "My Point"}
... }
>>> feature = await Feature.create_feature(layer, geojson)
"""
endpoint = urljoin(layer.endpoint, 'features/')
if srid != cls.BASE_SRID:
endpoint = f"{endpoint}?in_srid={srid}"
feature = await cls._create(layer.api, endpoint, geojson, factory_func=lambda api, item: AsyncFeature(layer, data=item))
feature.data['geometry'] = geojson['geometry']
feature._srid = srid
return feature
[docs]
@classmethod
async def get_feature(cls, layer: 'VectorLayer', feature_id: int, user_id: int = None) -> 'AsyncFeature':
"""
[async] Get a feature by its ID.
Args:
layer (VectorLayer): The vector layer the feature belongs to
feature_id (int): The ID of the feature
user_id (int): specific user. privileges required.
Returns:
AsyncFeature: The retrieved feature instance
Example:
>>> from geobox.aio import AsyncGeoboxClient
>>> from geobox.aio.feature import AsyncFeature
>>> async with AsyncGeoboxClient() as client:
>>> layer = await client.get_vector(uuid="12345678-1234-5678-1234-567812345678")
>>> feature = await AsyncFeature.get_feature(layer, feature_id=1)
"""
param = {
'f': 'json',
'user_id': user_id
}
endpoint = urljoin(layer.endpoint, f'features/')
return await cls._get_detail(layer.api, endpoint, uuid=feature_id, params=param, factory_func=lambda api, item: AsyncFeature(layer, data=item))
@property
def geometry(self) -> 'BaseGeometry':
"""
Get the feature geometry as a Shapely geometry object.
Returns:
shapely.geometry.BaseGeometry: The Shapely geometry object representing the feature's geometry
Raises:
ValueError: If the geometry is not a dictionary
ValueError: If the geometry type is not present in the feature data
ValueError: If the geometry coordinates are not present in the feature data
ImportError: If shapely is not installed
Example:
>>> from geobox.aio import AsyncGeoboxClient
>>> async with AsyncGeoboxClient() as client:
>>> layer = await client.get_vector(uuid="12345678-1234-5678-1234-567812345678")
>>> feature = await layer.get_feature(id=1)
>>> feature.geometry
"""
try:
from shapely.geometry import shape
except ImportError:
raise ImportError(
"The 'geometry' extra is required for this function. "
"Install it with: pip install geobox[geometry]"
)
if not self.data.get('geometry'):
raise ValueError("Geometry is not present in the feature data")
elif not isinstance(self.data['geometry'], dict):
raise ValueError("Geometry is not a dictionary")
elif not self.data['geometry'].get('type'):
raise ValueError("Geometry type is not present in the feature data")
elif not self.data['geometry'].get('coordinates'):
raise ValueError("Geometry coordinates are not present in the feature data")
else:
return shape(self.data['geometry'])
@geometry.setter
def geometry(self, value: object) -> None:
"""
Set the feature geometry.
Args:
value (object): The geometry to set.
Raises:
ValueError: If geometry type is not supported
ValueError: If the geometry has a different type than the layer type
ImportError: If shapely is not installed
Returns:
None
Example:
>>> from geobox.aio import AsyncGeoboxClient
>>> from shapely.affinity import translate
>>> async with AsyncGeoboxClient() as client:
>>> layer = await client.get_vector(uuid="12345678-1234-5678-1234-567812345678")
>>> feature = await layer.get_feature(id=1)
>>> geom = feature.geometry
>>> geom = translate(geom, 3.0, 0.5) # example change applied to the feature's geometry
>>> feature.geometry = geom
>>> await feature.save()
"""
try:
from shapely.geometry import mapping, Point, MultiPoint, LineString, MultiLineString, Polygon, MultiPolygon
except ImportError:
raise ImportError(
"The 'geometry' extra is required for this function. "
"Install it with: pip install geobox[geometry]"
)
if not isinstance(value, (Point, MultiPoint, LineString, MultiLineString, Polygon, MultiPolygon)):
raise ValueError("Geometry must be a Shapely geometry object")
elif self.feature_type and value.geom_type != self.feature_type.value:
raise ValueError("Geometry must have the same type as the layer type")
else:
self.data['geometry'] = mapping(value)
[docs]
async def _get_other_side_of_relationship(
self,
relationship: 'AsyncRelationship',
) -> Union['AsyncTable', 'AsyncVectorLayer']:
"""
[async] Determine which side of a relationship this table is on and return the opposite side.
Used internally to navigate bidirectional relationships.
Args:
relationship (AsyncRelationship): The relationship to examine.
Returns:
AsyncTable | AsyncVectorLayer: The endpoint (table or layer) on the opposite side
of the relationship from this table.
Raises:
ValueError: If this table is not part of the given relationship.
Note:
This method assumes the table is either the source or target,
not the relation table in Many-to-Many relationships.
"""
if relationship.source_id == self.layer.id:
return await relationship.get_target()
if relationship.target_id == self.layer.id:
return await relationship.get_source()
raise ValueError("Relationship does not involve this table.")
[docs]
async def associate_with(
self,
relationship_uuid: str,
*,
target_ids: Optional[List[int]] = None,
q: Optional[str] = None,
) -> Dict:
"""
[async] Create relationships between the source record and target records
Args:
relationship_uuid (str): the relationship uuid
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:
>>> layer = await client.get_vector(uuid="12345678-1234-5678-1234-567812345678")
>>> feature = await layer.get_feature(feature_id=1)
>>> await feature.associate_with(
... relationship_uuid="12345678-1234-5678-1234-567812345678",
... target_ids=[1, 2, 3],
... )
"""
relationship = await self.api.get_relationship(uuid=relationship_uuid)
return await relationship.associate_records(
source_id=self.id,
target_ids=target_ids,
q=q,
)
[docs]
async def disassociate_with(
self,
relationship_uuid: str,
*,
target_ids: Optional[List[int]] = None,
q: Optional[str] = None,
) -> Dict:
"""
[async] Remove relationships between the source record and target records
Args:
relationship_uuid (str): the relationship uuid
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 association result
Example:
>>> from geobox.aio import AsyncGeoboxClient
>>> async with AsyncGeoboxClient() as client:
>>> layer = await client.get_vector(uuid="12345678-1234-5678-1234-567812345678")
>>> feature = await layer.get_feature(feature_id=1)
>>> await feature.disassociate_with(
... relationship_uuid="12345678-1234-5678-1234-567812345678",
... target_ids=[1, 2, 3],
... )
"""
relationship = await self.api.get_relationship(uuid=relationship_uuid)
return await relationship.disassociate_records(
source_id=self.id,
target_ids=target_ids,
q=q,
)