Data Feeds API

Data Feeds API

Accessor: server.datafeeds

Reference: https://docs.tak.gov/api/takserver#tag/data-feed-api

Data feeds are server-side subscriptions that continuously pull CoT data matching a predicate (typically a CoT type filter) from a source. This API lets you manage predicate feeds, query the feed catalog (also by geographic bounds), read per-feed statistics and fetch the CoT messages stored in a feed.

The access-lockout trap — read this first

Warning

If you create a predicate feed with auth_type="X_509" and an empty filter_groups list, the server stores it in a state where nobody — not even the admin certificate — can read, update or delete it via the REST API. Every operation answers 403 Group access denied, and your only way out is direct database surgery.

Always pass at least one filter group that your identity belongs to, e.g. filter_groups=["WOLF_Friends"]. The build_predicate_feed() helper defaults to ["__ANON__"] for anonymous-accessible feeds; when in doubt, use a group you are actually a member of.

Also note: the server assigns its own UUID on creation. A uuid field in the create body is silently ignored.

And: minimal bodies ({"name": ...} alone) are rejected with HTTP 500. The server wants a fairly complete object — which is exactly what build_predicate_feed() produces.

---

build_predicate_feed()

def build_predicate_feed(
    name: str,
    predicate: str,
    predicate_lang: str = "JSON_PATH",
    source_endpoint: str | None = None,
    archive: bool = False,
    sync: bool = False,
    federated: bool = False,
    auth_type: str = "ANONYMOUS",
    filter_groups: list[str] | None = None,
    tags: list[str] | None = None,
) -> dict

Builds a complete PredicateDataFeed body with sane defaults. You will want to use this instead of hand-assembling the dictionary — it fills in the fields the server insists on and defaults filter_groups to ["__ANON__"] so the feed stays manageable after creation.

Parameter	Type	Description
name	str	Feed name
predicate	str	Filter expression, e.g. "type == 'a-u'"
predicate_lang	str	Predicate language, default "JSON_PATH"
source_endpoint	str or None	URL of the data source
archive / sync / federated	bool	Feed behaviour flags. archive defaults to False - archived feeds persist every CoT message in Postgres and quietly bloat the database; only enable it if you really intend to keep the data
auth_type	str	"ANONYMOUS", "X_509", ...
filter_groups	list[str] or None	Groups allowed to access the feed. Default ["__ANON__"]
tags	list[str] or None	Feed tags

Example:

feed_body = srv.datafeeds.build_predicate_feed(
    name="my-adsb-feed",
    predicate="type == 'a-f'",
    source_endpoint="https:// feeder.example.com:8080",
)
status, created = await srv.datafeeds.create_predicate_data_feed(feed_body)
print(created["uuid"])   # server-assigned UUID

---

get_data_feeds()

Usage: await server.datafeeds.get_data_feeds()

Returns all data feeds visible to the authenticated user.

API: GET /Marti/api/datafeeds

Returns: (status_code, data) — list of feed dicts.

get_data_feeds_in_bbox()

async def get_data_feeds_in_bbox(bbox: str) -> tuple[int, Any]

Feeds whose contents fall inside a bounding box. bbox format: "minLon,minLat,maxLon,maxLat".

API: GET /Marti/api/datafeeds/bounds/{bbox}

status, feeds = await srv.datafeeds.get_data_feeds_in_bbox("23.7,61.4,23.8,61.5")

get_data_feeds_in_polygon()

async def get_data_feeds_in_polygon(points: list[str]) -> tuple[int, Any]

Polygon variant. A curious endpoint: it is a GET with a JSON body (an array of point strings). The wrapper's connection layer passes JSON bodies on GETs, so this just works.

API: GET /Marti/api/datafeeds/bounds/polygon

status, feeds = await srv.datafeeds.get_data_feeds_in_polygon([
    "61.4,23.7", "61.5,23.7", "61.5,23.8", "61.4,23.8",
])

create_predicate_data_feed() / update_predicate_data_feed()

async def create_predicate_data_feed(feed: dict) -> tuple[int, Any]
async def update_predicate_data_feed(feed: dict, update_groups: bool = False) -> tuple[int, Any]

Create/update a predicate feed. The update takes the complete object including its uuid; update_groups=True additionally updates group assignments.

API: POST / PUT /Marti/api/datafeeds/predicate

Returns: (status_code, data) — the stored feed object.

get_predicate_data_feed() / delete_predicate_data_feed()

async def get_predicate_data_feed(feed_uuid: str) -> tuple[int, Any]
async def delete_predicate_data_feed(feed_guid: str) -> tuple[int, Any]

Fetch or remove one predicate feed by UUID/GUID.

get_stats() / get_stats_for_feed()

async def get_stats() -> tuple[int, Any]
async def get_stats_for_feed(uuid: str) -> tuple[int, Any]

Message statistics (message counts, rates, byte sizes, extents). Note that a freshly created feed has no messages yet — the per-feed endpoint then answers HTTP 500 with an envelope that carries no payload, which unwraps to None.

get_existing_cot_types() / get_cots_by_cot_type()

async def get_existing_cot_types(uuid: str) -> tuple[int, Any]
async def get_cots_by_cot_type(uuid: str, cot_type: str) -> tuple[int, Any]

Inspect the contents of a feed: first the list of CoT types currently present, then the actual stored CoT messages of one type.