OGC Features and Tiles API

app.py

"""OGC Features and Tiles API."""

import re

from fastapi import Depends, FastAPI, HTTPException, Path, Query

from starlette.requests import Request
from starlette.responses import Response, HTMLResponse
from starlette_cramjam.middleware import CompressionMiddleware

from morecantile import Tile, tms

from tifeatures.db import close_db_connection, connect_to_db, register_table_catalog
from tifeatures.factory import Endpoints as FeaturesEndpoints

from timvt.dependencies import  TileParams
from timvt.factory import TILE_RESPONSE_PARAMS, VectorTilerFactory, templates
from timvt.layer import Layer, Table
from timvt.models.mapbox import TileJSON

app = FastAPI(
    title="OGC Features and Tiles API",
    openapi_url="/api",
    docs_url="/api.html",
)
app.add_middleware(CompressionMiddleware)


# OGC Features.
endpoints = FeaturesEndpoints()
app.include_router(endpoints.router)


# OGC Tiles
def CollectionParam(
    request: Request,
    collectionId: str = Path(..., description="collection Name"),
) -> Layer:
    """Return Layer Object."""
    table_pattern = re.match(  # type: ignore
        r"^(?P<schema>.+)\.(?P<table>.+)$", collectionId
    )
    if not table_pattern:
        raise HTTPException(status_code=404, detail=f"Invalid Table format '{collectionId}'.")

    assert table_pattern.groupdict()["schema"]
    assert table_pattern.groupdict()["table"]

    table_catalog = getattr(request.app.state, "table_catalog", {})
    if collectionId in table_catalog:
        return Table(**table_catalog[collectionId])

    raise HTTPException(status_code=404, detail=f"Table '{collectionId}' not found.")


webmercator = tms.get("WebMercatorQuad")


@app.get("/collections/{collectionId}/tiles/{z}/{x}/{y}", **TILE_RESPONSE_PARAMS)
async def tile(
    request: Request,
    tile: Tile = Depends(TileParams),
    collection=Depends(CollectionParam),
):
    """Return vector tile."""
    pool = request.app.state.pool
    content = await collection.get_tile(pool, tile, webmercator)
    return Response(bytes(content), media_type="application/x-protobuf")


@app.get(
    "/collections/{collectionId}/tilejson.json",
    response_model=TileJSON,
    responses={200: {"description": "Return a tilejson"}},
    response_model_exclude_none=True,
)
async def tilejson(request: Request, collection=Depends(CollectionParam)):
    """Return TileJSON document."""
    path_params = {
        "collectionId": collection.id,
        "z": "{z}",
        "x": "{x}",
        "y": "{y}",
    }
    tile_endpoint = request.url_for("tile", **path_params)
    return {
        "minzoom": 0,
        "maxzoom": 22,
        "name": collection.id,
        "bounds": collection.bounds,
        "tiles": [tile_endpoint],
    }


@app.get("/collections/{collectionId}/viewer", response_class=HTMLResponse)
async def demo(request: Request, collection=Depends(CollectionParam)):
    tile_url = request.url_for("tilejson", collectionId=collection.id)
    return templates.TemplateResponse(
        name="viewer.html",
        context={"endpoint": tile_url, "request": request},
        media_type="text/html",
    )



@app.on_event("startup")
async def startup_event() -> None:
    """Connect to database on startup."""
    await connect_to_db(app)
    # TiMVT and TiFeatures share the same `Table_catalog` format
    # see https://github.com/developmentseed/timvt/pull/83
    await register_table_catalog(app)


@app.on_event("shutdown")
async def shutdown_event() -> None:
    """Close database connection."""
    await close_db_connection(app)