Skip to content

Instantly share code, notes, and snippets.

@thomaxxl
Last active January 20, 2021 09:46
Show Gist options
  • Save thomaxxl/53256daf0c94b53832fd6cc1959a1451 to your computer and use it in GitHub Desktop.
Save thomaxxl/53256daf0c94b53832fd6cc1959a1451 to your computer and use it in GitHub Desktop.
from flask import Flask
from flask_sqlalchemy import SQLAlchemy
from sqlalchemy import func
from geoalchemy2 import Geometry
from flask_sqlalchemy import SQLAlchemy
from safrs import SAFRSBase, SAFRSAPI
from flask.json import JSONEncoder
from flask import jsonify
import shapely_geojson
from safrs.json_encoder import SAFRSJSONEncoder
from geoalchemy2.functions import ST_AsGeoJSON
from geoalchemy2.elements import WKBElement, _SpatialElement
app = Flask(__name__)
db = SQLAlchemy()
class City(SAFRSBase, db.Model):
"""A city, including its geospatial data."""
__tablename__ = "cities"
point_id = db.Column(db.Integer, primary_key=True, autoincrement=True)
location = db.Column(db.String(30))
longitude = db.Column(db.Float)
latitude = db.Column(db.Float)
geo = db.Column(Geometry(geometry_type="POINT",srid=25833, dimension=2))
def __init__(self, *args, **kwargs):
print(args, kwargs)
latitude = kwargs.get("latitude",0)
longitude = kwargs.get("longitude",0)
kwargs['geo'] = 'POINT({} {})'.format(longitude, latitude)
super().__init__(*args,**kwargs)
def __repr__(self):
return "<City {name} ({lat}, {lon})>".format(
name=self.location, lat=self.latitude, lon=self.longitude)
def __str__(self):
return self.__repr__()
def get_cities_within_radius(self, radius):
"""Return all cities within a given radius (in meters) of this city."""
return City.query.filter(func.ST_Distance_Sphere(City.geo, self.geo) < radius).all()
@classmethod
def add_city(cls, location, longitude, latitude):
"""Put a new city in the database."""
geo = 'POINT({} {})'.format(longitude, latitude)
city = City(location=location,
longitude=longitude,
latitude=latitude,
geo=geo)
db.session.add(city)
db.session.commit()
@classmethod
def update_geometries(cls):
"""Using each city's longitude and latitude, add geometry data to db."""
cities = City.query.all()
for city in cities:
point = 'POINT({} {})'.format(city.longitude, city.latitude)
city.geo = point
db.session.commit()
def connect_to_db(app):
"""Connect the database to Flask app."""
app.config['SQLALCHEMY_DATABASE_URI'] = 'postgres:///testgis'
app.config['SQLALCHEMY_ECHO'] = False
app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = False
db.app = app
db.init_app(app)
def create_api(app, HOST="localhost", PORT=5000, API_PREFIX=""):
api = SAFRSAPI(app, host=HOST, port=PORT, prefix=API_PREFIX)
app.json_encoder = GeoJSONEncoder
api.expose_object(City)
print("Starting API: http://{}:{}/{}".format(HOST, PORT, API_PREFIX))
class GeoJSONEncoder(SAFRSJSONEncoder):
def default(self, obj, **kwargs):
if isinstance(obj, _SpatialElement):
return str(obj)
return super().default(obj, **kwargs)
if __name__ == "__main__":
connect_to_db(app)
db.create_all()
host = '192.168.235.136'
with app.app_context():
print(jsonify({}))
create_api(app, host)
for city in City.query.all():
print(city, city.location)
sf = db.session.query(City).filter(City.location == 'San Francisco').one()
app.run(host=host)
@ficapy
Copy link

ficapy commented Jan 20, 2021

👍

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