This is the official documentation of the forestadmin-agent-django and forestadmin-agent-flask Python agents.
Steps
Creating a custom data source will require you to work on the 3 following steps:
Declare the structure of the data
Declare the API capabilities
Code a translation layer
Minimal example
from typings import Optional
from forestadmin.agent_toolkit.utils.context import User
from forestadmin.datasource_toolkit.collections import Collection
from forestadmin.datasource_toolkit.datasources import Datasource
from forestadmin.datasource_toolkit.interfaces.query.aggregation import AggregateResult, Aggregation
from forestadmin.datasource_toolkit.interfaces.query.filter.paginated import PaginatedFilter
from forestadmin.datasource_toolkit.interfaces.query.filter.unpaginated import Filter
from forestadmin.datasource_toolkit.interfaces.query.projections import Projection
from forestadmin.datasource_toolkit.interfaces.records import RecordsDataAlias
import requests # client for the target API
# The real work is in writing this module
# Expect a full featured query translation module to be over 1000 LOCs
from .forest_query_translation import QueryGenerator
# Minimal implementation of a readonly data source
class MyCollection(Collection):
def __init__(self, datasource):
# Set name of the collection once imported
super().__init__("MyCollection", datasource)
# structure
self.add_field("id", {
"type": "Column",
"column_type": "Number",
"is_primary_key": True,
"is_read_only": True, # field is readonly
# As we are using the query translation strategy => define capabilities
"filter_operators": set(), # field is not filterable
"is_sortable": False, # field is not sortable
})
self.add_field("title", {
"type": "Column",
"column_type": "String",
"is_read_only": True,
"filter_operators": set(),
"is_sortable": False,
})
async def list(
self,
caller: User,
filter_: PaginatedFilter,
projection: Projection
) -> List[RecordsDataAlias]:
params = QueryGenerator.generate_list_query_string(filter_, projection)
response = requests.get('https://my-api/my-collection', params)
return response.json()["items"]
async def aggregate(
self,
caller: User,
filter_: Filter,
aggregation: Aggregation,
limit: Optional[int]
) -> List[AggregateResult]:
params = QueryGenerator.generate_aggregate_query_string(
filter_,
aggregation,
limit
)
response = requests.get('https://my-api/my-collection', params)
return response.json()["items"]
class MyDatasource(Datasource):
def __init__(self):
super().__init__()
self.add_collection(MyCollection(self))
from custom_datasources.my_datasource import MyDatasource
agent.add_datasource(MyDatasource())
Read more
Implementing a data source using the "query translation" strategy is an advanced concept: you will need to have a deep understanding of Forest Admin internals.
This strategy is a good match when writing data sources for full-featured databases.
Before starting, it is highly advised to read and understand the following sections:
