find_points test added.

This commit is contained in:
Kian 2024-08-22 18:09:12 -04:00
parent 158d83341b
commit 0c07d7ea93
3 changed files with 28 additions and 11 deletions

Binary file not shown.

Binary file not shown.

View File

@ -1,23 +1,40 @@
import os
from flask import jsonify
from sqlalchemy import inspect
from flask import jsonify, request
from sqlalchemy import inspect, func, text
from sqlalchemy.orm import sessionmaker
from geoalchemy2 import Geometry
from shapely import wkb
def register_route(app,db):
@app.route("/")
def hello_world():
return "<p>Hello, World!</p>"
@app.route('/user/')
@app.route('/user/<username>')
def show_user_profile(username=None):
test = os.getenv("MY_KEY")
if username:
return f"<p>Hello, Mr {username}! You are the man.</p> {test}"
return "<p>Hello, user!</p>"
@app.route('/table')
def list_table():
inspector = inspect(db.engine)
tables = inspector.get_table_names()
return jsonify(tables)
@app.route('/find_points/', methods=['POST'])
def find_points():
data = request.json
coordinates = data.get('coordinates', [])
polygon_wkt = 'POLYGON(({}))'.format(', '.join(f'{lat} {lng}' for lat, lng in coordinates))
# Construct the query string
query = f"""
SELECT id, geom FROM agents
WHERE ST_Contains(
ST_Transform(ST_GeomFromText('{polygon_wkt}', 4326), 4326),
geom
)
"""
with db.engine.connect() as connection:
results = connection.execute(text(query)).fetchall()
# Convert results to a list of dictionaries for JSON serialization
results_list = [{"id": result.id, "geom": [wkb.loads(bytes.fromhex(result.geom)).x, wkb.loads(bytes.fromhex(result.geom)).y]} for result in results]
return jsonify(results_list)