Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def test_get_col_spec(self):
g = Geography(srid=900913)
assert g.get_col_spec() == 'geography(GEOMETRY,900913)'
{'query': sqltypes.TEXT, 'date_type': data_types.TEXT},
{'query': sqltypes.BOOLEAN, 'date_type': data_types.BOOLEAN},
{'query': sqltypes.INTEGER, 'date_type': data_types.INTEGER},
{'query': sqltypes.SMALLINT, 'date_type': data_types.INTEGER},
{'query': sqltypes.BIGINT, 'date_type': data_types.INTEGER},
{'query': sqltypes.NUMERIC, 'date_type': data_types.FLOAT},
{'query': sqltypes.DATETIME, 'date_type': data_types.DATE_TIME},
{'query': sqltypes.TIMESTAMP, 'date_type': data_types.DATE_TIME},
{'query': sqltypes.JSON, 'date_type': data_types.JSON},
]
default_data_type = data_types.TEXT
try:
from geoalchemy2 import types
map_data_types.append({'query': types.Geometry, 'date_type': data_types.GEOMETRY})
map_data_types.append({'query': types.Geography, 'date_type': data_types.GEOGRAPHY})
except ImportError:
pass
def map_data_type(value):
for rule in reversed(map_data_types):
if isinstance(value, rule['query']):
return rule['date_type']
return default_data_type
'__callback__': lambda f: {
'__default_type__': geotypes.Geography if f.geography else geotypes.Geometry,
'__default_type_kwargs__': {
'geometry_type': f.geom_type,
'srid': f.srid,
'dimension': f.dim,
'spatial_index': f.spatial_index,
},
'Construct a MultiPolygon given an arbitrary collection of closed '
'linestrings as a MultiLineString text representation Well-Known text '
'representation.'),
('ST_Box2dFromGeoHash', types.Geometry,
'Return a BOX2D from a GeoHash string.'),
('ST_GeogFromText', types.Geography,
'Return a specified geography value from Well-Known Text representation '
'or extended (WKT).'),
('ST_GeographyFromText', types.Geography,
'Return a specified geography value from Well-Known Text representation '
'or extended (WKT).'),
('ST_GeogFromWKB', types.Geography,
'Creates a geography instance from a Well-Known Binary geometry '
'representation (WKB) or extended Well Known Binary (EWKB).'),
('ST_GeomFromTWKB', types.Geometry,
'Creates a geometry instance from a TWKB ("Tiny Well-Known Binary") '
'geometry representation.'),
('ST_GeomCollFromText', types.Geometry,
'Makes a collection Geometry from collection WKT with the given SRID. If '
'SRID is not given, it defaults to 0.'),
('ST_GeomFromEWKB', types.Geometry,
'Return a specified ST_Geometry value from Extended Well-Known Binary '
'representation (EWKB).'),
('ST_GeomFromEWKT', types.Geometry,
class GeometryDump(CompositeType):
"""
The return type for functions like ``ST_Dump``, consisting of a path and
a geom field. You should normally never use this class directly.
"""
typemap = {'path': postgresql.ARRAY(Integer), 'geom': Geometry}
""" Dictionary defining the contents of a ``geometry_dump``. """
# Register Geometry, Geography and Raster to SQLAlchemy's Postgres reflection
# subsystem.
ischema_names['geometry'] = Geometry
ischema_names['geography'] = Geography
ischema_names['raster'] = Raster
{'query': sqltypes.VARCHAR, 'date_type': data_types.TEXT},
{'query': sqltypes.TEXT, 'date_type': data_types.TEXT},
{'query': sqltypes.BOOLEAN, 'date_type': data_types.BOOLEAN},
{'query': sqltypes.INTEGER, 'date_type': data_types.INTEGER},
{'query': sqltypes.SMALLINT, 'date_type': data_types.INTEGER},
{'query': sqltypes.NUMERIC, 'date_type': data_types.FLOAT},
{'query': sqltypes.DATETIME, 'date_type': data_types.DATE_TIME},
{'query': sqltypes.TIMESTAMP, 'date_type': data_types.DATE_TIME},
{'query': sqltypes.JSON, 'date_type': data_types.JSON},
]
default_data_type = data_types.TEXT
try:
from geoalchemy2 import types
map_data_types.append({'query': types.Geometry, 'date_type': data_types.GEOMETRY})
map_data_types.append({'query': types.Geography, 'date_type': data_types.GEOGRAPHY})
except ImportError:
pass
def map_data_type(value):
for rule in reversed(map_data_types):
if isinstance(value, rule['query']):
return rule['date_type']
return default_data_type
def refine_with_user_area(query):
"""Takes a query and refines it with a spatial constraint
based on user setting"""
if 'lon' and 'lat' and 'radius' in session:
return query.filter(ST_DWithin(
cast(Task.location, Geography),
cast(from_shape(Point(session["lon"], session["lat"])), Geography),
session["radius"]))
else:
return query
# sqlalchemy.FloatField: {'filter_class': NumberFilter},
# sqlalchemy.NullBooleanField: {'filter_class': BooleanFilter},
# sqlalchemy.SlugField: {'filter_class': CharFilter},
# sqlalchemy.EmailField: {'filter_class': CharFilter},
# sqlalchemy.FilePathField: {'filter_class': CharFilter},
# sqlalchemy.URLField: {'filter_class': CharFilter},
# sqlalchemy.GenericIPAddressField: {'filter_class': CharFilter},
# sqlalchemy.CommaSeparatedIntegerField: {'filter_class': CharFilter},
# sqlalchemy.UUIDField: {'filter_class': UUIDFilter}
}
FILTER_FOR_DBFIELD_DEFAULT = FILTER_FOR_DBFIELD[sqltypes.VARCHAR]
try:
from geoalchemy2 import types
FILTER_FOR_DBFIELD[types.Geometry] = {'filter_class': WKTFilter, 'lookups': geography_lookups}
FILTER_FOR_DBFIELD[types.Geography] = {'filter_class': WKTFilter, 'lookups': geography_lookups}
except ImportError:
pass
def filter_for_data_type(value):
for date_type, filter_data in FILTER_FOR_DBFIELD.items():
if isinstance(value, date_type):
return filter_data
return FILTER_FOR_DBFIELD_DEFAULT
def refine_with_user_area(query):
"""Takes a query and refines it with a spatial constraint
based on user setting"""
if 'lon' and 'lat' and 'radius' in session:
return query.filter(ST_DWithin(
cast(Task.location, Geography),
cast(from_shape(Point(session["lon"], session["lat"])), Geography),
session["radius"]))
else:
return query
def generate_geom_filters(geometry, Pois):
filters, geom = [], None
if 'bbox' in geometry and 'geom' not in geometry:
geom = geometry['bbox'].wkt
filters.append(
geo_func.ST_DWithin(geo_func.ST_Buffer(type_coerce(geom, Geography), geometry['buffer']), Pois.geom, 0))
elif 'bbox' in geometry and 'geom' in geometry:
geom_bbox = geometry['bbox'].wkt
geom = geometry['geom'].wkt
filters.append( # in bbox
geo_func.ST_DWithin(
geo_func.ST_Intersection(geo_func.ST_Buffer(type_coerce(geom, Geography), geometry['buffer']),
type_coerce(geom_bbox, Geography)), Pois.geom, 0))
elif 'bbox' not in geometry and 'geom' in geometry:
geom = geometry['geom'].wkt
filters.append( # buffer around geom
geo_func.ST_DWithin(geo_func.ST_Buffer(type_coerce(geom, Geography), geometry['buffer']), Pois.geom, 0)
)
return filters, geom