Python Server Proxy

For more idioms see ../integrations/python.md.

Flask

import os, time, hmac, hashlib
from flask import Flask, Response
import requests

app = Flask(__name__)
KEY    = os.environ['MFD_SRV_KEY']
SECRET = os.environ['MFD_SRV_SECRET']
BASE   = 'https://tiles.mapsfordevs.com'

def sign(method, path):
    ts = str(int(time.time()))
    sig = hmac.new(SECRET.encode(), f'{method}\n{path}\n{ts}'.encode(), hashlib.sha256).hexdigest()
    return {'Authorization': f'Bearer {KEY}', 'X-MFD-Timestamp': ts, 'X-MFD-Signature': sig}

@app.get('/tiles/<int:z>/<int:x>/<int:y>.pbf')
def tile(z, x, y):
    path = f'/tiles/{z}/{x}/{y}.pbf'
    r = requests.get(f'{BASE}{path}', headers=sign('GET', path), stream=True, timeout=10)
    return Response(
        r.iter_content(8192),
        status=r.status_code,
        content_type='application/x-protobuf',
        headers={'Cache-Control': 'public, max-age=86400'}
    )

FastAPI (async)

import os, time, hmac, hashlib
from fastapi import FastAPI
from fastapi.responses import StreamingResponse
import httpx

app   = FastAPI()
KEY    = os.environ['MFD_SRV_KEY']
SECRET = os.environ['MFD_SRV_SECRET']
client = httpx.AsyncClient(base_url='https://tiles.mapsfordevs.com', timeout=10)

def sign(method: str, path: str) -> dict[str, str]:
    ts = str(int(time.time()))
    sig = hmac.new(SECRET.encode(), f'{method}\n{path}\n{ts}'.encode(), hashlib.sha256).hexdigest()
    return {'Authorization': f'Bearer {KEY}', 'X-MFD-Timestamp': ts, 'X-MFD-Signature': sig}

@app.get('/tiles/{z}/{x}/{y}.pbf')
async def tile(z: int, x: int, y: int):
    path = f'/tiles/{z}/{x}/{y}.pbf'
    r = await client.get(path, headers=sign('GET', path))
    return StreamingResponse(
        r.aiter_bytes(),
        status_code=r.status_code,
        media_type='application/x-protobuf',
        headers={'Cache-Control': 'public, max-age=86400'}
    )

Django

# views.py
import os, time, hmac, hashlib
from django.http import HttpResponse
from django.views.decorators.cache import cache_page
import requests

KEY    = os.environ['MFD_SRV_KEY']
SECRET = os.environ['MFD_SRV_SECRET']

def sign(method, path):
    ts = str(int(time.time()))
    sig = hmac.new(SECRET.encode(), f'{method}\n{path}\n{ts}'.encode(), hashlib.sha256).hexdigest()
    return {'Authorization': f'Bearer {KEY}', 'X-MFD-Timestamp': ts, 'X-MFD-Signature': sig}

@cache_page(60 * 60 * 24)
def tile(request, z, x, y):
    path = f'/tiles/{z}/{x}/{y}.pbf'
    r = requests.get(f'https://tiles.mapsfordevs.com{path}', headers=sign('GET', path), timeout=10)
    return HttpResponse(r.content, status=r.status_code, content_type='application/x-protobuf')

# urls.py
# path('tiles/<int:z>/<int:x>/<int:y>.pbf', views.tile)

Redis-backed tile cache

import redis
r_cache = redis.Redis(host='localhost', port=6379, decode_responses=False)

@app.get('/tiles/{z}/{x}/{y}.pbf')
async def tile(z: int, x: int, y: int):
    path = f'/tiles/{z}/{x}/{y}.pbf'
    cached = r_cache.get(path)
    if cached:
        return Response(cached, media_type='application/x-protobuf', headers={'X-Cache': 'HIT'})
    r = await client.get(path, headers=sign('GET', path))
    if r.status_code == 200:
        r_cache.setex(path, 86400, r.content)
    return Response(r.content, status_code=r.status_code,
                    media_type='application/x-protobuf',
                    headers={'X-Cache': 'MISS'})

ASGI deployment

Use uvicorn with multiple workers + uvloop:

uvicorn main:app --host 0.0.0.0 --port 8080 --workers 4 --loop uvloop