Skip to content

Instantly share code, notes, and snippets.

@magopian

magopian/aiohttp + aiopg.py

Last active October 4, 2019 08:48
Show Gist options
  • Save magopian/30615bb181b8747037bc71972c85ff07 to your computer and use it in GitHub Desktop.
Save magopian/30615bb181b8747037bc71972c85ff07 to your computer and use it in GitHub Desktop.
Download ZIP
experiments with python asyncio (all three are roughly the same speed as the werkzeug+pg version, as it's CPU bound, not IO bound: asyncio doesn't help in such cases)
Raw
aiohttp + aiopg.py
"""start with:
gunicorn utilery.views_aiohttp_aiopg:app -b 0.0.0.0:3579 -w 4 -k aiohttp.worker.GunicornWebWorker"""
import asyncio
import json
import math
import time
import psycopg2
import psycopg2.extras
from werkzeug.exceptions import BadRequest, HTTPException, abort
from werkzeug.routing import Map, Rule
from werkzeug.wrappers import Request, Response
import queries
import mercantile
import mapbox_vector_tile
from aiohttp import web
import aiopg
def to_geometry():
return ('ST_AsText(ST_TransScale(%s, %.12f, %.12f, %.12f, %.12f)) as _way' # noqa
% (self.GEOMETRY, -self.west, -self.south,
self.SCALE / (self.east - self.west),
self.SCALE / (self.north - self.south)))
def to_layer(name, features):
return {
"name": name,
"features": features
}
def to_feature(row):
return {
"geometry": row['_way'],
"properties": row_to_dict(row)
}
def row_to_dict(row):
def f(item):
return not item[0].startswith('_') and item[0] != 'way'
return dict(i for i in row.items() if f(i))
async def handle(request):
layers = []
for name, query in queries.QUERIES.items():
async with request.app['pool'].acquire() as conn:
async with conn.cursor(cursor_factory=psycopg2.extras.DictCursor) as cur:
await cur.execute(query)
rows = await cur.fetchall()
features = [to_feature(row) for row in rows]
layers.append(to_layer(name, features))
output = mapbox_vector_tile.encode(layers)
return web.Response(body=output)
async def setup_db():
return await aiopg.create_pool('dbname=utilery user=mathieu password= host=localhost')
app = web.Application()
app.router.add_route('GET', '/', handle)
loop = asyncio.get_event_loop()
pool = loop.run_until_complete(setup_db())
app['pool'] = pool
Raw
falcon + asyncpg.py
"""Start with:
gunicorn utilery.views_falcon_asyncpg:app -b 0.0.0.0:3579 -w 4
"""
import time
import asyncio
import asyncpg
import queries
import mapbox_vector_tile
import falcon
def to_layer(name, features):
return {
"name": name,
"features": features
}
def to_feature(row, name):
return {
"geometry": row['_way'],
"properties": row_to_dict(row)
}
def row_to_dict(row):
def f(item):
return not item[0].startswith('_') and item[0] != 'way'
return dict(i for i in row.items() if f(i))
async def query_to_layer(query, name):
async with pool.acquire() as conn:
rows = await conn.fetch(query)
# features = []
# async for row in cur:
# features.append(to_feature(row, name))
features = [to_feature(row, name) for row in rows]
return to_layer(name, features)
class Foo:
def on_get(self, req, resp):
tasks = [query_to_layer(query, name) for name, query in queries.QUERIES.items()]
loop = asyncio.get_event_loop()
layers = loop.run_until_complete(asyncio.gather(*tasks))
output = mapbox_vector_tile.encode(layers)
resp.body = output
resp.content_type = 'application/x-protobuf'
async def setup_db():
return await asyncpg.create_pool(user='mathieu', database='utilery')
loop = asyncio.get_event_loop()
pool = loop.run_until_complete(setup_db())
app = falcon.API()
foo = Foo()
app.add_route('/', foo)
Raw
falcon + pg.py
"""Start with:
gunicorn utilery.views_falcon_pg:app -b 0.0.0.0:3579 -w 4
"""
import json
import math
import time
import psycopg2
import psycopg2.extras
from werkzeug.exceptions import BadRequest, HTTPException, abort
from werkzeug.routing import Map, Rule
from werkzeug.wrappers import Request, Response
from . import config
from .core import DB, RECIPES
from .plugins import Plugins
import queries
import mercantile
import mapbox_vector_tile
import falcon
connection = psycopg2.connect('dbname=utilery user=mathieu password= host=localhost')
def to_geometry():
return ('ST_AsText(ST_TransScale(%s, %.12f, %.12f, %.12f, %.12f)) as _way' # noqa
% (self.GEOMETRY, -self.west, -self.south,
self.SCALE / (self.east - self.west),
self.SCALE / (self.north - self.south)))
def to_layer(name, features):
return {
"name": name,
"features": features
}
def to_feature(row):
return {
"geometry": row['_way'],
"properties": row_to_dict(row)
}
def row_to_dict(row):
def f(item):
return not item[0].startswith('_') and item[0] != 'way'
return dict(i for i in row.items() if f(i))
class Foo:
def on_get(self, req, resp):
layers = []
for name, query in queries.QUERIES.items():
cur = connection.cursor(cursor_factory=psycopg2.extras.DictCursor)
cur.execute(query)
rows = cur.fetchall()
cur.close()
features = [to_feature(row) for row in rows]
layers.append(to_layer(name, features))
output = mapbox_vector_tile.encode(layers)
resp.body = output
resp.content_type = 'application/x-protobuf'
app = falcon.API()
foo = Foo()
app.add_route('/', foo)
Raw
queries.py
QUERIES = {
'railways': """SELECT ST_AsText(ST_TransScale(ST_Intersection(way, ST_Expand(ST_SetSRID(ST_MakeBox2D(ST_MakePoint(5009377.085697311, -1448023.0638343792), ST_MakePoint(5028944.964938316, -1428455.184593374)), 3857), 611.49622628141)), -5009377.085697310977, 1448023.063834379194, 0.209322632747, 0.209322632747)) as _way, * FROM (SELECT
ST_SimplifyPreserveTopology(ST_Union(ST_buffer(geometry,20,'join=mitre')),76.43702828517625/2) AS way, 'railway' as type, ST_Area(ST_Union(ST_buffer(geometry,20))) as area
FROM
osm_railways
WHERE
service in ('yard','spur','siding') and geometry && ST_Expand(ST_SetSRID(ST_MakeBox2D(ST_MakePoint(5009377.085697311, -1448023.0638343792), ST_MakePoint(5028944.964938316, -1428455.184593374)), 3857), 611.49622628141)
UNION
SELECT
ST_SimplifyPreserveTopology(ST_Union(ST_buffer(b.geometry,50,'join=mitre mitre_limit=1')),76.43702828517625/2) AS way, 'buildings' as type, ST_Area(ST_Union(ST_buffer(b.geometry,50))) as area
FROM
osm_buildings b
LEFT JOIN osm_landusages l ON (ST_Intersects(l.geometry, b.geometry) and l.type in ('residential','retail','industrial','farmyard'))
WHERE
b.geometry && ST_Expand(ST_SetSRID(ST_MakeBox2D(ST_MakePoint(5009377.085697311, -1448023.0638343792), ST_MakePoint(5028944.964938316, -1428455.184593374)), 3857), 611.49622628141) and b.area>50 and l.type is null) AS data WHERE ST_IsValid(way) AND ST_Intersects(way, ST_Expand(ST_SetSRID(ST_MakeBox2D(ST_MakePoint(5009377.085697311, -1448023.0638343792), ST_MakePoint(5028944.964938316, -1428455.184593374)), 3857), 611.49622628141))
""",
'places': """SELECT ST_AsText(ST_TransScale(way, -5009377.085697310977, 1448023.063834379194, 0.209322632747, 0.209322632747)) as _way, * FROM (SELECT
name, type, osm_id, geometry AS way
FROM
osm_places
WHERE
name IS NOT NULL
AND type IN ('city','county','province','island','town','lake','village')
ORDER BY
z_order ASC) AS data WHERE ST_IsValid(way) AND ST_Intersects(way, ST_Expand(ST_SetSRID(ST_MakeBox2D(ST_MakePoint(5009377.085697311, -1448023.0638343792), ST_MakePoint(5028944.964938316, -1428455.184593374)), 3857), 9783.93962050256))
""",
'roads': """SELECT ST_AsText(ST_TransScale(ST_Intersection(way, ST_Expand(ST_SetSRID(ST_MakeBox2D(ST_MakePoint(5009377.085697311, -1448023.0638343792), ST_MakePoint(5028944.964938316, -1428455.184593374)), 3857), -152.8740565703525)), -5009377.085697310977, 1448023.063834379194, 0.209322632747, 0.209322632747)) as _way, * FROM (SELECT
ST_LineMerge(ST_Collect(geometry)) AS way,
type, name, ref, tunnel, bridge,
(CASE WHEN type IN ('motorway', 'motorway_link') THEN 'highway'
WHEN type IN ('trunk', 'trunk_link', 'secondary', 'primary') THEN 'major_road'
ELSE 'minor_road' END) AS category
FROM
osm_roads_gen1
WHERE
geometry && ST_Expand(ST_SetSRID(ST_MakeBox2D(ST_MakePoint(5009377.085697311, -1448023.0638343792), ST_MakePoint(5028944.964938316, -1428455.184593374)), 3857), -152.8740565703525) AND name != ''
GROUP BY
type, name, ref, tunnel, bridge,category,z_order
ORDER BY
z_order ASC, st_length(ST_LineMerge(ST_Collect(geometry))) ASC) AS data WHERE ST_IsValid(way) AND ST_Intersects(way, ST_Expand(ST_SetSRID(ST_MakeBox2D(ST_MakePoint(5009377.085697311, -1448023.0638343792), ST_MakePoint(5028944.964938316, -1428455.184593374)), 3857), -152.8740565703525))""",
'waterways': """SELECT ST_AsText(ST_TransScale(ST_Intersection(way, ST_Expand(ST_SetSRID(ST_MakeBox2D(ST_MakePoint(5009377.085697311, -1448023.0638343792), ST_MakePoint(5028944.964938316, -1428455.184593374)), 3857), 611.49622628141)), -5009377.085697310977, 1448023.063834379194, 0.209322632747, 0.209322632747)) as _way, * FROM (SELECT
geometry AS way, type, name, tunnel, bridge
FROM
osm_waterways_gen1
WHERE
type in ('river','canal')) AS data WHERE ST_IsValid(way) AND ST_Intersects(way, ST_Expand(ST_SetSRID(ST_MakeBox2D(ST_MakePoint(5009377.085697311, -1448023.0638343792), ST_MakePoint(5028944.964938316, -1428455.184593374)), 3857), 611.49622628141)) """,
'landusages': """SELECT ST_AsText(ST_TransScale(way, -5009377.085697310977, 1448023.063834379194, 0.209322632747, 0.209322632747)) as _way, * FROM (SELECT
ST_PointOnSurface(ST_Union(geometry)) AS way,
round(log(sum(area))::numeric,0)::int as area_log,
length(name) as name_length,
type, name
FROM
osm_landusages_gen1
WHERE
geometry && ST_Expand(ST_SetSRID(ST_MakeBox2D(ST_MakePoint(5009377.085697311, -1448023.0638343792), ST_MakePoint(5028944.964938316, -1428455.184593374)), 3857), 9783.93962050256) and name !='' and area>(100*76.43702828517625*76.43702828517625)
GROUP BY
type,name,z_order
ORDER BY
z_order DESC, sum(area) DESC) AS data WHERE ST_IsValid(way) AND ST_Intersects(way, ST_Expand(ST_SetSRID(ST_MakeBox2D(ST_MakePoint(5009377.085697311, -1448023.0638343792), ST_MakePoint(5028944.964938316, -1428455.184593374)), 3857), 9783.93962050256)) """,
'railways_gen1': """
SELECT ST_AsText(ST_TransScale(ST_Intersection(way, ST_Expand(ST_SetSRID(ST_MakeBox2D(ST_MakePoint(5009377.085697311, -1448023.0638343792), ST_MakePoint(5028944.964938316, -1428455.184593374)), 3857), 611.49622628141)), -5009377.085697310977, 1448023.063834379194, 0.209322632747, 0.209322632747)) as _way, * FROM (SELECT
ST_SimplifyPreserveTopology(ST_LineMerge(ST_Collect(geometry)),76.43702828517625) AS way,
type, name, tunnel, bridge, oneway, ref, service, highspeed, class
FROM
osm_railways_gen1
WHERE
geometry && ST_Expand(ST_SetSRID(ST_MakeBox2D(ST_MakePoint(5009377.085697311, -1448023.0638343792), ST_MakePoint(5028944.964938316, -1428455.184593374)), 3857), 611.49622628141) and type in ('rail','subway') and coalesce(service,'') in ('','main')
GROUP BY
z_order, type, name, tunnel, bridge, oneway, ref, service, highspeed, class
ORDER BY
z_order, highspeed DESC, coalesce(service,'') DESC) AS data WHERE ST_IsValid(way) AND ST_Intersects(way, ST_Expand(ST_SetSRID(ST_MakeBox2D(ST_MakePoint(5009377.085697311, -1448023.0638343792), ST_MakePoint(5028944.964938316, -1428455.184593374)), 3857), 611.49622628141))""",
'roads_gen1': """
SELECT ST_AsText(ST_TransScale(ST_Intersection(way, ST_Expand(ST_SetSRID(ST_MakeBox2D(ST_MakePoint(5009377.085697311, -1448023.0638343792), ST_MakePoint(5028944.964938316, -1428455.184593374)), 3857), 1222.99245256282)), -5009377.085697310977, 1448023.063834379194, 0.209322632747, 0.209322632747)) as _way, * FROM (SELECT
geometry AS way, type, name, tunnel, bridge,
(CASE WHEN type IN ('motorway', 'motorway_link') THEN 'highway'
WHEN type IN ('trunk', 'trunk_link', 'secondary', 'primary') THEN 'major_road'
ELSE 'minor_road' END) AS category,
(CASE WHEN type LIKE '%_link' THEN 1
ELSE 0 END) AS link
FROM
osm_roads_gen1
ORDER BY
z_order ASC) AS data WHERE ST_IsValid(way) AND ST_Intersects(way, ST_Expand(ST_SetSRID(ST_MakeBox2D(ST_MakePoint(5009377.085697311, -1448023.0638343792), ST_MakePoint(5028944.964938316, -1428455.184593374)), 3857), 1222.99245256282))""",
'waterareas_gen1': """
SELECT ST_AsText(ST_TransScale(ST_Intersection(way, ST_Expand(ST_SetSRID(ST_MakeBox2D(ST_MakePoint(5009377.085697311, -1448023.0638343792), ST_MakePoint(5028944.964938316, -1428455.184593374)), 3857), 611.49622628141)), -5009377.085697310977, 1448023.063834379194, 0.209322632747, 0.209322632747)) as _way, * FROM (SELECT
geometry AS way, type, name
FROM
osm_waterareas_gen1
WHERE
area > (76.43702828517625*76.43702828517625*4)
ORDER BY
area DESC) AS data WHERE ST_IsValid(way) AND ST_Intersects(way, ST_Expand(ST_SetSRID(ST_MakeBox2D(ST_MakePoint(5009377.085697311, -1448023.0638343792), ST_MakePoint(5028944.964938316, -1428455.184593374)), 3857), 611.49622628141))""",
'landusages_gen1': """
SELECT ST_AsText(ST_TransScale(ST_Intersection(way, ST_Expand(ST_SetSRID(ST_MakeBox2D(ST_MakePoint(5009377.085697311, -1448023.0638343792), ST_MakePoint(5028944.964938316, -1428455.184593374)), 3857), 611.49622628141)), -5009377.085697310977, 1448023.063834379194, 0.209322632747, 0.209322632747)) as _way, * FROM (SELECT
ST_SimplifyPreserveTopology(geometry,76.43702828517625) AS way, type, name
FROM
osm_landusages_gen1
WHERE
geometry && ST_Expand(ST_SetSRID(ST_MakeBox2D(ST_MakePoint(5009377.085697311, -1448023.0638343792), ST_MakePoint(5028944.964938316, -1428455.184593374)), 3857), 611.49622628141) and area>(76.43702828517625*76.43702828517625*4)
ORDER BY
z_order DESC, area DESC) AS data WHERE ST_IsValid(way) AND ST_Intersects(way, ST_Expand(ST_SetSRID(ST_MakeBox2D(ST_MakePoint(5009377.085697311, -1448023.0638343792), ST_MakePoint(5028944.964938316, -1428455.184593374)), 3857), 611.49622628141))"""
}
Raw
werkzeug + aiopg.py
"""start with:
gunicorn utilery.views_werkzeug_aiopg:app -b 0.0.0.0:3579 -w 4
"""
import json
import math
import time
import psycopg2
import psycopg2.extras
from werkzeug.exceptions import BadRequest, HTTPException, abort
from werkzeug.routing import Map, Rule
from werkzeug.wrappers import Request, Response
import queries
import mercantile
import mapbox_vector_tile
import aiopg
connection = psycopg2.connect('dbname=utilery user=mathieu password= host=localhost')
def to_geometry():
return ('ST_AsText(ST_TransScale(%s, %.12f, %.12f, %.12f, %.12f)) as _way' # noqa
% (self.GEOMETRY, -self.west, -self.south,
self.SCALE / (self.east - self.west),
self.SCALE / (self.north - self.south)))
def to_layer(name, features):
return {
"name": name,
"features": features
}
def to_feature(row):
return {
"geometry": row['_way'],
"properties": row_to_dict(row)
}
def row_to_dict(row):
def f(item):
return not item[0].startswith('_') and item[0] != 'way'
return dict(i for i in row.items() if f(i))
@Request.application
def app(request):
layers = []
conn = await aiopg.connect('dbname=utilery user=mathieu password= host=localhost')
for name, query in queries.QUERIES.items():
async with conn.cursor(cursor_factory=psycopg2.extras.DictCursor) as cur:
await cur.execute(query)
rows = cur.fetchall()
features = [to_feature(row) for row in rows]
layers.append(to_layer(name, features))
output = mapbox_vector_tile.encode(layers)
return Response(output)
Raw
werkzeug + pg.py
import json
import math
import time
import psycopg2
import psycopg2.extras
from werkzeug.exceptions import BadRequest, HTTPException, abort
from werkzeug.routing import Map, Rule
from werkzeug.wrappers import Request, Response
from . import config
from .core import DB, RECIPES
from .plugins import Plugins
import queries
import mercantile
import mapbox_vector_tile
connection = psycopg2.connect('dbname=utilery user=mathieu password= host=localhost')
def to_geometry():
return ('ST_AsText(ST_TransScale(%s, %.12f, %.12f, %.12f, %.12f)) as _way' # noqa
% (self.GEOMETRY, -self.west, -self.south,
self.SCALE / (self.east - self.west),
self.SCALE / (self.north - self.south)))
def to_layer(name, features):
return {
"name": name,
"features": features
}
def to_feature(row):
return {
"geometry": row['_way'],
"properties": row_to_dict(row)
}
def row_to_dict(row):
def f(item):
return not item[0].startswith('_') and item[0] != 'way'
return dict(i for i in row.items() if f(i))
@Request.application
def app(request):
layers = []
for name, query in queries.QUERIES.items():
cur = connection.cursor(cursor_factory=psycopg2.extras.DictCursor)
cur.execute(query)
rows = cur.fetchall()
cur.close()
features = [to_feature(row) for row in rows]
layers.append(to_layer(name, features))
output = mapbox_vector_tile.encode(layers)
return Response(output)
@yohanboniface
Copy link 

"""Start with:
gunicorn utilery.views_falcon_pg:app -b 0.0.0.0:3579 -w 4
"""
import time

import asyncio
import asyncpg

import queries

import mapbox_vector_tile
import falcon


def to_layer(name, features):
    return {
        "name": name,
        "features": features
    }


def to_feature(row, name):
    return {
        "geometry": row['_way'],
        "properties": row_to_dict(row)
    }


def row_to_dict(row):
    def f(item):
        return not item[0].startswith('_') and item[0] != 'way'
    return dict(i for i in row.items() if f(i))


async def query_to_layer(query, name):
    db = await asyncpg.connect(user='ybon', database='utilery')
    rows = await db.fetch(query)
    # features = []
    # async for row in cur:
    #     features.append(to_feature(row, name))

    features = [to_feature(row, name) for row in rows]
    await db.close()
    return to_layer(name, features)


class Foo:
    def on_get(self, req, resp):
        tasks = []
        for name, query in queries.QUERIES.items():
            future = asyncio.ensure_future(query_to_layer(query, name))
            tasks.append(future)
                # tasks = [query_to_layer(cur, query, name) for name, query in queries.QUERIES.items()]
        loop = asyncio.get_event_loop()
        layers = loop.run_until_complete(asyncio.gather(*tasks))

        output = mapbox_vector_tile.encode(layers)
        resp.body = output
        resp.content_type = 'application/x-protobuf'

app = falcon.API()
foo = Foo()
app.add_route('/', foo)

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment