Avoid making a second network call from inside Lambda?

[MattReimer]

I did search for a way to do this but maybe it was looking me right in the face and I didn't see it. I'm fairly new to fastAPI so that may be it.

Here is a simplified version of my Lambda function. I have removed a lot of functionality so it's easy to talk about:

import json
import os
from fastapi import FastAPI, Request
from fastapi.responses import Response
from mangum import Mangum
from titiler.core.factory import TilerFactory
from titiler.application.settings import ApiSettings
from titiler.core.errors import DEFAULT_STATUS_CODES, add_exception_handlers
import httpx

api_settings = ApiSettings()
app = FastAPI(title="Riverscapes COG tiler", description="Cloud Optimized GeoTIFF")
cog = TilerFactory()
app.include_router(cog.router, tags=["Cloud Optimized GeoTIFF"])

add_exception_handlers(app, DEFAULT_STATUS_CODES)

BUCKET_NAME = os.environ.get("BUCKET_NAME", "DUMMY")


@app.get("/{z}/{x}/{y}.png")
async def get_tile(z: int, x: int, y: int, request: Request):
    """_summary_

    Args:
        z (int): _description_
        x (int): _description_
        y (int): _description_
        request (Request): _description_

    Returns:
        _type_: _description_
    """
    query_params = dict(request.query_params)
    rsxpath = query_params.get("file")

    # Modify the query parameters as needed to fit Titiler's expected format
    query_params = getattr(request.state, "query_params", {})
    colormap = {
        0: [0, 0, 0, 0],
        1: [255, 0, 0, 255],
    }
    query_params['url'] = f"s3://{BUCKET_NAME}/cogs/{rsxpath}/COG.tiff"
    query_params["colormap"] = json.dumps(colormap)
    query_params.pop("file", None)
    query_params.pop("symbology", None)

    # Now that we have our query params set we pass this to Titiler to get the actual tile
    async with httpx.AsyncClient() as client:
        response = await client.get(f"http://127.0.0.1:8007/tiles/WebMercatorQuad/{z}/{x}/{y}.png", params=query_params)
        return Response(content=response.content, media_type=response.headers.get("content-type", "application/octet-stream"))


# Here is the actual Lambda Handler
handler = Mangum(app, api_gateway_base_path=api_settings.root_path, lifespan="auto")

My issue is this:

On lambda you can't call "127.0.0.1" and it seems to me that making a network call from inside another network call is unnecessary.

I can't find a way to call titiler's tile() function directly though and avoid using httpx to make a second call

I have to be missing something obvious like cog.tile(x=x, y=y, z=z) or something right?

[vincentsarago]

are you using ☝️ just because you want to pass file=your file instead of the full url? if yes you can create a custom path_dependency such as:

from typing import Annotated
from fastapi import Query

from titiler.core.factory import TilerFactory

def DatasetPathParams(file: Annotated[str, Query(description="File Name")]) -> str:
    """Create dataset path from args"""
    return f"s3://{BUCKET_NAME}/cogs/{file}/COG.tiff"


endpoints = TileFactory(path_dependency= DatasetPathParams)

If it's to have a default colormap, you could also have a custom colormap dependency

import json
from typing import Literal, Optional, Sequence
from fastapi import Query
from rio_tiler.colormap import cmap, parse_color
from titiler.core.factory import TilerFactory

def ColorMapDependency(
    colormap_name: Annotated[  # type: ignore
        Literal[tuple(cmap.list())],
        Query(description="Colormap name"),
    ] = None,
    colormap: Annotated[
        Optional[str], Query(description="JSON encoded custom Colormap")
    ] = None,
):
    if colormap_name:
        return cmap.get(colormap_name)

    elif colormap:
        try:
            c = json.loads(
                colormap,
                object_hook=lambda x: {
                    int(k): parse_color(v) for k, v in x.items()
                },
            )

            # Make sure to match colormap type
            if isinstance(c, Sequence):
                c = [(tuple(inter), parse_color(v)) for (inter, v) in c]

            return c
        except json.JSONDecodeError as e:
            raise HTTPException(
                status_code=400, detail="Could not parse the colormap value."
            ) from e
    
    return {
        0: [0, 0, 0, 0],
        1: [255, 0, 0, 255],
    }

endpoints = TilerFactory(colormap_dependency=ColorMapDependency)

you could also easily copy/paste https://github.com/developmentseed/titiler/blob/main/src/titiler/core/titiler/core/factory.py#L751C1-L838C1

[MattReimer]

Huh. That's really food for thought. Thank you. I think I've been trying to reinvent a couple of wheels here.

[MattReimer]

Followup:

What happens if my ColorMapDependency needs information (like the max and min pixel values) from the statistics endpoint?

Can I fetch the statistics just using Titiler without needing to make an httpx call?

EDIT: I think I just answered my own question from your answer above.

I'm going to try copying and pasting

https://github.com/developmentseed/titiler/blob/main/src/titiler/core/titiler/core/factory.py#L437-L451

into my code.

[MattReimer]

I am soooooooooo close to having this working. On your advice I refactored my code to use FastAPI dependency injection but I can't figure out how to use pre-existing titiler dependencies to fetch the max and min values:

from typing import Annotated, Optional, Sequence, Dict
import os
import json
from fastapi import FastAPI, HTTPException, Depends
from mangum import Mangum
from titiler.core.factory import TilerFactory
from titiler.application.settings import ApiSettings
from titiler.core.errors import DEFAULT_STATUS_CODES, add_exception_handlers


BUCKET_NAME = os.environ.get("BUCKET_NAME", "DUMMY")


class TileCommonProps:
    """This is just a simple class to manage all the common context"""
    minRasterVal: int = 0
    maxRasterVal: int = 255
    colormap: Optional[Dict[int, Sequence[int]]] = None
    rsxpath: str = ""
    symbology: str = ""
    project_type: str = ""
    ramp_type: str = "INTERPOLATED"
    s3_path: str = ""

    def __init__(self, rsxpath: str, project_type: str, symbology: str, colormap: Optional[Dict[int, Sequence[int]]] = None,
                 minRasterVal: int = 0, maxRasterVal: int = 255, ramp_type: str = "INTERPOLATED"):
        self.rsxpath = rsxpath
        self.project_type = project_type
        self.symbology = symbology
        self.colormap = colormap
        self.minRasterVal = minRasterVal
        self.maxRasterVal = maxRasterVal
        self.ramp_type = ramp_type
        self.s3_path = f"s3://{BUCKET_NAME}/cogs/{project_type}/{rsxpath}/COG.tiff"


def common_parameters(rsx, pt, s) -> TileCommonProps:
    """This is all dependency injection for FastAPI"""

    # =================================================================
    # RIGHT HERE I NEED THE Min and max values for the raster
    # =================================================================
    minRasterVal, maxRasterVal = ???

    if s:
        try:
            # NOTE: I removed this to keep the example short
            colormap, minRasterVal, maxRasterVal, ramp_type = get_colormap(pt, s, [minRasterVal, maxRasterVal])
            return TileCommonProps(
                rsxpath=rsx,
                project_type=pt,
                symbology=s,
                colormap=colormap,
                minRasterVal=minRasterVal,
                maxRasterVal=maxRasterVal,
                ramp_type=ramp_type,
            )
        except json.JSONDecodeError as e:
            raise HTTPException(
                status_code=400, detail="Could not parse the colormap value."
            ) from e
    return TileCommonProps(
        rsxpath=rsx,
        project_type=pt,
        symbology=s,
    )


CommonsDep = Annotated[dict, Depends(common_parameters)]


def DatasetPathParamsDependency(common: CommonsDep) -> str:
    return common.s3_path


def ColorMapDependency(common: CommonsDep) -> Optional[Dict]:
    return common.colormap


api_settings = ApiSettings()
cog = TilerFactory(
    path_dependency=DatasetPathParamsDependency,
    colormap_dependency=ColorMapDependency,
)

# Define the FastAPI App
app = FastAPI(title="Riverscapes COG tiler", description="Cloud Optimized GeoTIFF")
app.include_router(cog.router, tags=["Cloud Optimized GeoTIFF"])
add_exception_handlers(app, DEFAULT_STATUS_CODES)


# Here is the actual Lambda Handler
handler = Mangum(app, api_gateway_base_path=api_settings.root_path, lifespan="auto")


def get_colormap(pt, s, maxmin):
    # NOTE: I removed this function to keep the example short. Pretend it returns something interesting
    # and uses the max and min raster values to rescale the colormap
    return [], 0, 255, "INTERPOLATED"

It feels like I'm a single line away from success but I can't see it.

[MattReimer]

Ok. I think I figured it out. Here's my final pattern. I left out the qgis_to_titiler_colormap to keep the example short.

from typing import Annotated, Optional, Sequence, Tuple, Dict
import os
import logging
import time
import httpx
from fastapi import FastAPI, Depends
from mangum import Mangum
from rio_tiler.io import Reader
from titiler.core.factory import TilerFactory
from titiler.core.dependencies import ImageRenderingParams
from titiler.application.settings import ApiSettings
from titiler.core.errors import DEFAULT_STATUS_CODES, add_exception_handlers



BUCKET_NAME = os.environ.get("BUCKET_NAME", "DUMMY")
# Set this to False in an env file to disable caching for testing
USE_CACHE = os.environ.get("SKIP_CACHE", "True").lower() == "true"
# Hardcode where we get our symbology data from
RIVERSCAPES_XML_URL = "https://SOMEURL.net"
# Manifest URL for web symbologies
MANIFEST_URL = f"{RIVERSCAPES_XML_URL}/web_raster_symbologies.txt"
CACHE_DURATION = 30  # seconds
print('USE_CACHE', USE_CACHE)


class CacheValue:
    """ CacheValue: Just a really simple class to hold a value and a timestamp
    """

    def __init__(self, value):
        self.value = value
        self.timestamp = time.time()

    def __getitem__(self, key):
        return self.value[key]

    def expired(self) -> bool:
        """Expired

        Returns:
            _type_: _description_
        """
        return (time.time() - self.timestamp) > CACHE_DURATION


# In-memory cache for symbology data
CACHE: Dict[str, CacheValue] = {}


class TileCommonProps:
    """This is just a simple class to manage all the common context"""
    rsxpath: str = ""
    symbology: str = ""
    project_type: str = ""
    s3_path: str = ""

    def __init__(self, rsxpath: str, project_type: str, symbology: str):
        self.rsxpath = rsxpath
        self.project_type = project_type
        self.symbology = symbology
        self.s3_path = f"s3://{BUCKET_NAME}/cogs/{project_type}/{rsxpath}/COG.tiff"

    def get_min_max(self) -> Tuple[int, int]:
        """Fetch the min and max values of the raster"""
        cache_key = f"minmax::{self.s3_path}"
        if cache_key in CACHE and not CACHE[cache_key].expired() and USE_CACHE:
            print(f"CACHE HIT minmax: {CACHE[cache_key].value}")
            return CACHE[cache_key].value
        else:
            with Reader(self.s3_path) as cog_reader:
                stats = cog_reader.statistics()
            minRasterVal = stats['b1'].min
            maxRasterVal = stats['b1'].max
            ret_val = (minRasterVal, maxRasterVal)
            print(f"CACHE MISS minmax: {ret_val}")
            CACHE[cache_key] = CacheValue(ret_val)
            return ret_val

    def get_manifest(self) -> Sequence[str]:
        """Fetch the manifest of available symbologies from Riverscapes XML"""
        # https://xml.riverscapes.net/web_raster_symbologies.txt
        manifest_cache_key = "manifest"
        manifest = []
        if manifest_cache_key in CACHE and not CACHE[manifest_cache_key].expired() and USE_CACHE:
            print("CACHE HIT manifest")
            manifest = CACHE[manifest_cache_key].value
        else:
            with httpx.Client() as client:
                response = client.get(MANIFEST_URL)
                symbology_manifest_cache = response.text.split("\n")
                print("CACHE MISS manifest")
                CACHE[manifest_cache_key] = CacheValue(symbology_manifest_cache)
                manifest = symbology_manifest_cache
        return manifest

    def get_colormap(self) -> Dict[int, Sequence[int]]:
        """Fetch the symbology from Riverscapes XML"""
        manifest = self.get_manifest()
        minRasterVal, maxRasterVal = self.get_min_max()

        final_url = 'Shared/hillshade'
        shared_url = f"Shared/{self.symbology}"
        project_url = f"{self.project_type}/{self.symbology}"

        if project_url in manifest:
            final_url = project_url
        elif shared_url in manifest:
            final_url = shared_url

        symbology_cache_key = f"S::{final_url}"
        if symbology_cache_key in CACHE and not CACHE[symbology_cache_key].expired() and USE_CACHE:
            print(f"CACHE HIT symbology: {final_url}")
            return CACHE[symbology_cache_key].value

        symb_url = f"{RIVERSCAPES_XML_URL}/Symbology/web/{final_url}.txt"  # Replace with the actual URL
        with httpx.Client() as client:
            response = client.get(symb_url)
            response.raise_for_status()
        # The fallback for the symbology is the index file so we just throw if that's what happens
        if '.json' in response.text:
            raise ValueError(f"No valid symbology found at url: {symb_url}")

        colormap = qgis_to_titiler_colormap(response.text, [minRasterVal, maxRasterVal])
        print(f"CACHE MISS symbology: {final_url}")
        CACHE[symbology_cache_key] = CacheValue(colormap)
        return colormap


def common_parameters(rsx, pt, s) -> TileCommonProps:
    """This is all dependency injection for FastAPI. It's a bit of a mess but it works.
    """
    props = TileCommonProps(
        rsxpath=rsx,
        project_type=pt,
        symbology=s,
    )
    return props


CommonsDep = Annotated[TileCommonProps, Depends(common_parameters)]


def DatasetPathParamsDependency(common: CommonsDep) -> str:
    """Rewire the input parameters of the COG tiler to match the Riverscapes XML URL structure"""
    return common.s3_path


def ColorMapDependency(common: CommonsDep) -> Optional[Dict]:
    """ Change the input parameters of the COG tiler to match the Riverscapes XML URL structure"""
    return common.get_colormap()[0]


def RenderDependency(common: CommonsDep) -> ImageRenderingParams:
    """ Change the input parameters of the COG tiler to match the Riverscapes XML URL structure"""
    minRasterVal, maxRasterVal = common.get_min_max()
    colormap = common.get_colormap()
    rescale = None
    if colormap[3] == "INTERPOLATED":
        rescale = [f'{minRasterVal},{maxRasterVal}']
    elif colormap[3] == "DISCRETE":
        rescale = [f'{colormap[1]},{colormap[2]}']
    elif colormap[3] == "EXACT":
        rescale = None
    return ImageRenderingParams(
        rescale=rescale,
    )

api_settings = ApiSettings()
cog = TilerFactory(
    path_dependency=DatasetPathParamsDependency,
    colormap_dependency=ColorMapDependency,
    render_dependency=RenderDependency,
)

# Define the FastAPI App
app = FastAPI(title="Riverscapes COG tiler", description="Cloud Optimized GeoTIFF")
app.include_router(cog.router, tags=["Cloud Optimized GeoTIFF"])
add_exception_handlers(app, DEFAULT_STATUS_CODES)


# Here is the actual Lambda Handler
handler = Mangum(app, api_gateway_base_path=api_settings.root_path, lifespan="auto")