是否有用于 Microsoft SQL 服务器 Geography/Geometry 类型的 SQLAlchemy UserDefinedType 示例？
Is there an example SQLAlchemy UserDefinedType for Microsoft SQL Server Geography/Geometry Types?
如何创建一个 SQLAlchemy UserDefinedType 允许我插入 SQL 服务器上的地理数据类型？
我正在使用 Python 3.6 和 Pandas to_sql to write into a SQL Server table that will have a column with a geography data type. I am using SQLAlchemy.create_engine 创建到 SQLExpress 的数据库连接，使用 DRIVER={ODBC Driver 13 for SQL Server}。我将多边形存储在 GeoPandas 数据框中。
SQLAlchemy 1.3.10 does not directly provide support for Geography or Geometry data types; and, GeoAlchemy2 does not support MS SQL Server. I have been trying to use SQLAlchemy's UserDefinedType 看看我是否能得到结果：
GEOGRAPHY::STGeomFromText(GEOMETRY::STGeomFromText('POLYGON((-110.17315242968752 52.66767554218751,-110.18536282187502 52.66770066015627,-110.19718901640618 52.66771763203127,-110.197593865625 52.667718411718795,-110.19747227656248 52.67594785000003,-110.1732282007812 52.67592660234379,-110.17315242968752 52.66767554218751))',4269).MakeValid().STUnion(GEOMETRY::STGeomFromText('POLYGON((-110.17315242968752 52.66767554218751,-110.18536282187502 52.66770066015627,-110.19718901640618 52.66771763203127,-110.197593865625 52.667718411718795,-110.19747227656248 52.67594785000003,-110.1732282007812 52.67592660234379,-110.17315242968752 52.66767554218751))',4269).STStartPoint()).STAsText(),4269)
到目前为止我有：
class Geography(UserDefinedType):
def get_col_spec(self):
return "GEOGRAPHY"
def bind_processor(self, dialect):
def process(value):
if value is None:
return None
return 'GEOGRAPHY::STGeomFromText(GEOMETRY::STGeomFromText({0},4269).MakeValid().STUnion(GEOMETRY::STGeomFromText({0},4269).STStartPoint()).STAsText(),4269)'.format("'" + value + "'")
return process
虽然我被这个结果困住了（请注意整个事情周围的引号。:
"GEOGRAPHY::STGeomFromText(GEOMETRY::STGeomFromText('POLYGON ((-110.1731524296875 52.66767554218751, -110.185362821875 52.6677006601563, -110.19718901 ... (382 characters truncated) ... -110.1974722765625 52.67594785, -110.1732282007812 52.67592660234379, -110.1731524296875 52.66767554218751))',4269).STStartPoint()).STAsText(),4269)"
我知道这会起作用 SQL:
DECLARE @g NVARCHAR(MAX)
SELECT @g = 'POLYGON((-110.17315242968752 52.66767554218751,-110.18536282187502 52.66770066015627,-110.19718901640618 52.66771763203127,-110.197593865625 52.667718411718795,-110.19747227656248 52.67594785000003,-110.1732282007812 52.67592660234379,-110.17315242968752 52.66767554218751))'
INSERT INTO dbo.[Spatial_Table] ([geometry])
VALUES (GEOGRAPHY::STGeomFromText(GEOMETRY::STGeomFromText(@g,4269).MakeValid().STUnion(GEOMETRY::STGeomFromText(@g,4269).STStartPoint()).STAsText(),4269))
使用 pandas.DataFrame.to_sql 导致错误：
DataError: (pyodbc.DataError) ('22018', '[22018] [Microsoft][ODBC Driver 13 for SQL Server][SQL Server]Operand type clash: ntext is incompatible with geography (206) (SQLExecDirectW); [22018] [Microsoft][ODBC Driver 13 for SQL Server][SQL Server]Statement(s) could not be prepared. (8180)')
[SQL: INSERT INTO dbo.[Spatial_Table] (geometry) VALUES (?)]
[parameters: (("GEOGRAPHY::STGeomFromText(GEOMETRY::STGeomFromText('POLYGON ((-114.4742908039062 51.94055257031255, -114.4623163671875 51.94054674921875, -114.450628 ... (3922 characters truncated) ... .4978060703125 51.9479693648438, -114.4743004117187 51.94797611093753, -114.4742908039062 51.94055257031255))',4269).STStartPoint()).STAsText(),4269)",), ("GEOGRAPHY::STGeomFromText(GEOMETRY::STGeomFromText('POLYGON ((-112.1157004828125 49.53477394218754, -112.1156657703125 49.52339953203125, -112.124386 ... (704 characters truncated) ... 1212683476562 49.53115973984376, -112.1212804195312 49.53477826953127, -112.1157004828125 49.53477394218754))',4269).STStartPoint()).STAsText(),4269)",))]
(Background on this error at: http://sqlalche.me/e/9h9h)
我假设错误是由于周围的引号导致了一个字符串，而不是传递给 SQL 的函数。
我尝试在 bind_processor return 字符串上使用 sqlalchemy.sql.expression.text 但我收到此错误：
ProgrammingError: (pyodbc.ProgrammingError) ('Invalid parameter type. param-index=0 param-type=TextClause', 'HY105')
[SQL: INSERT INTO [Spatial_Table] (geometry) VALUES (?)]
[parameters: (<sqlalchemy.sql.elements.TextClause object at 0x000002566D5A40F0>,)]
(Background on this error at: http://sqlalche.me/e/f405)
而且我看过sqlalchemy.sql.expression.func；但是，我不知道如何将它与复杂的 GEOGRAPHY::STGeomFromText 方法一起使用。
虽然我的推测可能不正确，因为这将适用于 SQL:
INSERT INTO dbo.[Spatial_Table] ([geometry])
VALUES ('POLYGON((-110.17315242968752 52.66767554218751,-110.18536282187502 52.66770066015627,-110.19718901640618 52.66771763203127,-110.197593865625 52.667718411718795,-110.19747227656248 52.67594785000003,-110.1732282007812 52.67592660234379,-110.17315242968752 52.66767554218751))')
只需使用：
class Geography(sqlalchemy.types.UserDefinedType):
def get_col_spec(self):
return "GEOGRAPHY"
将导致此错误，即使参数看起来是正确的。它仍然出现相同的操作数类型冲突错误：
DataError: (pyodbc.DataError) ('22018', '[22018] [Microsoft][ODBC Driver 13 for SQL Server][SQL Server]Operand type clash: ntext is incompatible with geography (206) (SQLExecDirectW); [22018] [Microsoft][ODBC Driver 13 for SQL Server][SQL Server]Statement(s) could not be prepared. (8180)')
[SQL: INSERT INTO [Spatial_Table] (geometry) VALUES (?)]
[parameters: ('POLYGON ((-114.4742908039062 51.94055257031255, -114.4623163671875 51.94054674921875, -114.4506284421875 51.94053819687502, -114.4506230351562 51.933 ... (1739 characters truncated) ... -114.49781949375 51.93693783750001, -114.4978060703125 51.9479693648438, -114.4743004117187 51.94797611093753, -114.4742908039062 51.94055257031255))',)]
(Background on this error at: http://sqlalche.me/e/9h9h)
这是一个完整的例子：
import geopandas
import pyodbc
import urllib
import sqlalchemy
params = 'DRIVER={ODBC Driver 13 for SQL Server};' \
'SERVER=ServerName;' \
'PORT=1433;' \
'DATABASE=DatabaseName;' \
'trusted_connection=yes;'
params = urllib.parse.quote_plus(params)
db = sqlalchemy.create_engine('mssql+pyodbc:///?odbc_connect=%s' % params)
class Geography(sqlalchemy.types.UserDefinedType):
def get_col_spec(self):
return "GEOGRAPHY"
def bind_processor(self, dialect):
def process(value):
if value is None:
return None
return 'GEOGRAPHY::STGeomFromText(GEOMETRY::STGeomFromText({0},4269).MakeValid().STUnion(GEOMETRY::STGeomFromText({0},4269).STStartPoint()).STAsText(),4269)'.format("'" + value + "'")
return process
gdf = geopandas.GeoDataFrame({'geometry':['POLYGON ((-114.4742908039062 51.94055257031255, -114.4623163671875 51.94054674921875, -114.4506284421875 51.94053819687502, -114.4506230351562 51.93329010000002, -114.4506172351562 51.92605222890626, -114.4270012117187 51.92605734296876, -114.42699041875 51.91881815312502, -114.4269793164062 51.91139827421875, -114.4150635234375 51.91142824140627, -114.4150633671875 51.90417951171878, -114.41506316875 51.89693553984375, -114.403441165625 51.89697914921879, -114.3914711632812 51.89696492578128, -114.3914339117187 51.88973628046875, -114.3913953804687 51.88226671015627, -114.3677831210937 51.88229111953126, -114.3677989414062 51.87505186875001, -114.3678146921875 51.86781865390628, -114.3678270578125 51.86058223671876, -114.3678396773437 51.8531650226563, -114.3561627734375 51.85316232265626, -114.34419459375 51.85315451953124, -114.34420228125 51.8459314015625, -114.3442099304687 51.83869999843751, -114.3561678164062 51.83869954843755, -114.3561729164063 51.83145872656252, -114.3561489476562 51.82404763359375, -114.3678250734375 51.82404504843754, -114.367827496875 51.82784562734378, -114.3678343414063 51.83869510312502, -114.3797993828125 51.83869063437504, -114.3797996945312 51.84592788906252, -114.3798041078125 51.85316721640629, -114.3914338414062 51.85316111093755, -114.3914139953125 51.86780720156253, -114.4033583140625 51.86778718046878, -114.4150253679687 51.86777429218751, -114.415020278125 51.8822367257813, -114.432806471875 51.88222315703126, -114.4505780710937 51.882230434375, -114.450589053125 51.88963689218753, -114.450594271875 51.89689434218752, -114.4622547703125 51.89689832265628, -114.4622642132812 51.90414635312504, -114.4622736234375 51.91139415781254, -114.4742338164062 51.91140009453125, -114.4742472351562 51.91883031875, -114.4742673734375 51.92607473984378, -114.4858996257812 51.92606931250003, -114.4978195835938 51.92606766953128, -114.49781949375 51.93693783750001, -114.4978060703125 51.9479693648438, -114.4743004117187 51.94797611093753, -114.4742908039062 51.94055257031255))']})
gdf.to_sql('Spatial_Table',
if_exists='replace',
index=False,
dtype={'geometry': Geography},
con=db)
我希望在 SQLAlchemy 或 GeoAlchemy 中看到直接 support for Geography and Geometry for MS SQL Server。
而不是在 Python 中进行值处理的 bind_processor()，使用 bind_expression() 进行 SQL 端处理：
class Geography(sqlalchemy.types.UserDefinedType):
def get_col_spec(self):
return "GEOGRAPHY"
def bind_expression(self, bindvalue):
# Note that this does *not* format the value to the expression text, but
# the bind value key.
return text(f'GEOGRAPHY::STGeomFromText(GEOMETRY::STGeomFromText(:{bindvalue.key},4269).MakeValid().STUnion(GEOMETRY::STGeomFromText(:{bindvalue.key},4269).STStartPoint()).STAsText(),4269)').bindparams(bindvalue)
我刚遇到同样的问题并找到了解决方案。这可能有帮助：
from sqlalchemy import func
from sqlalchemy.sql.expression import text
from sqlalchemy.sql.schema import Column
from sqlalchemy.types import UserDefinedType, TypeEngine
from geojson_funcs import wkt2geojson, geojson2wkt
class Geography(UserDefinedType):
cache_ok = True
def __init__(self, srid: int = 4326):
self.srid = srid
def get_col_spec(self):
return "GEOGRAPHY"
def bind_expression(self, bindvalue):
# not able to use func here since "::" in function name
exp = f"geography::STGeomFromText(:{bindvalue.key}, {self.srid})"
f = text(exp)
return f.bindparams(bindvalue)
def column_expression(self, col: Column):
# not able to use func here since function needs to be applied as method to the column itself
# col.func() not func(col)
exp = f"{col.key}.STAsText()"
f1 = text(exp)
f2 = func.IIF(col == None, None, f1, type_=self)
return f2
def bind_processor(self, dialect):
def process(value):
return geojson2wkt(value) if value else value
return process
def result_processor(self, dialect, coltype):
def process(value):
return wkt2geojson(value) if value else value
return process
class comparator_factory(TypeEngine.Comparator):
def __add__(self, other):
return self.op("goofy")(other)
def intersects_bounds(self, lon_min, lon_max, lat_min, lat_max):
p = f'POLYGON (({lon_min} {lat_min}, {lon_max} {lat_min}, {lon_max} {lat_max}, {lon_min} {lat_max}, {lon_min} {lat_min}))'
exp = f"geography::STGeomFromText('{p}', {self.type.srid}).STIntersects({self.expr.name}) = 1"
f = text(exp)
return f
我唯一的问题是代码中的注释：
- bind_expression: 由于函数名称中的“::”
，无法在此处使用 func
- column_expression：无法在此处使用 func，因为需要将函数作为方法应用于列本身 -> col.func() not func (col)
如果有人知道如何解决这些问题，我很乐意在这里更新我的示例:)
此外，我将此函数用于 return GeoJson 而不是 WKT：
from typing import Union
import geojson
import shapely.wkt
from shapely.geometry import shape, GeometryCollection, Point, MultiPoint, \
LineString, MultiLineString, Polygon, MultiPolygon
GeoJsonType: type = Union[geojson.Feature, geojson.FeatureCollection]
WktType: type = Union[GeometryCollection, Point, MultiPoint, LineString, MultiLineString, Polygon, MultiPolygon]
def wkt2geojson(wkt: str) -> GeoJsonType:
wkt = shapely.wkt.loads(wkt)
wkt = validate_wkt(wkt)
if type(wkt) == GeometryCollection:
features = [geojson.Feature(geometry=o, properties={}) for o in wkt.geoms]
feature_collection = geojson.FeatureCollection(features)
return feature_collection
else:
feature = geojson.Feature(geometry=wkt, properties={})
return feature
def geojson2wkt(data: Union[GeoJsonType, dict]) -> WktType:
data = validate_geojson(data)
if data.get('type') == 'Feature':
g = shape(data.get('geometry'))
return g.wkt
elif data.get('type') == 'FeatureCollection':
g = [shape(f.get('geometry')) for f in data.get('features')]
gc = GeometryCollection(g)
return gc.wkt
else:
raise Exception
def validate_geojson(data: Union[GeoJsonType, dict]) -> GeoJsonType:
data = geojson.loads(geojson.dumps(data))
if not data.is_valid:
raise ValueError('value is not a valid GeoJson')
return data
def validate_wkt(data: WktType) -> WktType:
if not data.is_valid:
raise ValueError('value is not a valid WKT')
return data
如何创建一个 SQLAlchemy UserDefinedType 允许我插入 SQL 服务器上的地理数据类型？
我正在使用 Python 3.6 和 Pandas to_sql to write into a SQL Server table that will have a column with a geography data type. I am using SQLAlchemy.create_engine 创建到 SQLExpress 的数据库连接，使用 DRIVER={ODBC Driver 13 for SQL Server}。我将多边形存储在 GeoPandas 数据框中。
SQLAlchemy 1.3.10 does not directly provide support for Geography or Geometry data types; and, GeoAlchemy2 does not support MS SQL Server. I have been trying to use SQLAlchemy's UserDefinedType 看看我是否能得到结果：
GEOGRAPHY::STGeomFromText(GEOMETRY::STGeomFromText('POLYGON((-110.17315242968752 52.66767554218751,-110.18536282187502 52.66770066015627,-110.19718901640618 52.66771763203127,-110.197593865625 52.667718411718795,-110.19747227656248 52.67594785000003,-110.1732282007812 52.67592660234379,-110.17315242968752 52.66767554218751))',4269).MakeValid().STUnion(GEOMETRY::STGeomFromText('POLYGON((-110.17315242968752 52.66767554218751,-110.18536282187502 52.66770066015627,-110.19718901640618 52.66771763203127,-110.197593865625 52.667718411718795,-110.19747227656248 52.67594785000003,-110.1732282007812 52.67592660234379,-110.17315242968752 52.66767554218751))',4269).STStartPoint()).STAsText(),4269)
到目前为止我有：
class Geography(UserDefinedType):
def get_col_spec(self):
return "GEOGRAPHY"
def bind_processor(self, dialect):
def process(value):
if value is None:
return None
return 'GEOGRAPHY::STGeomFromText(GEOMETRY::STGeomFromText({0},4269).MakeValid().STUnion(GEOMETRY::STGeomFromText({0},4269).STStartPoint()).STAsText(),4269)'.format("'" + value + "'")
return process
虽然我被这个结果困住了（请注意整个事情周围的引号。:
"GEOGRAPHY::STGeomFromText(GEOMETRY::STGeomFromText('POLYGON ((-110.1731524296875 52.66767554218751, -110.185362821875 52.6677006601563, -110.19718901 ... (382 characters truncated) ... -110.1974722765625 52.67594785, -110.1732282007812 52.67592660234379, -110.1731524296875 52.66767554218751))',4269).STStartPoint()).STAsText(),4269)"
我知道这会起作用 SQL:
DECLARE @g NVARCHAR(MAX)
SELECT @g = 'POLYGON((-110.17315242968752 52.66767554218751,-110.18536282187502 52.66770066015627,-110.19718901640618 52.66771763203127,-110.197593865625 52.667718411718795,-110.19747227656248 52.67594785000003,-110.1732282007812 52.67592660234379,-110.17315242968752 52.66767554218751))'
INSERT INTO dbo.[Spatial_Table] ([geometry])
VALUES (GEOGRAPHY::STGeomFromText(GEOMETRY::STGeomFromText(@g,4269).MakeValid().STUnion(GEOMETRY::STGeomFromText(@g,4269).STStartPoint()).STAsText(),4269))
使用 pandas.DataFrame.to_sql 导致错误：
DataError: (pyodbc.DataError) ('22018', '[22018] [Microsoft][ODBC Driver 13 for SQL Server][SQL Server]Operand type clash: ntext is incompatible with geography (206) (SQLExecDirectW); [22018] [Microsoft][ODBC Driver 13 for SQL Server][SQL Server]Statement(s) could not be prepared. (8180)')
[SQL: INSERT INTO dbo.[Spatial_Table] (geometry) VALUES (?)]
[parameters: (("GEOGRAPHY::STGeomFromText(GEOMETRY::STGeomFromText('POLYGON ((-114.4742908039062 51.94055257031255, -114.4623163671875 51.94054674921875, -114.450628 ... (3922 characters truncated) ... .4978060703125 51.9479693648438, -114.4743004117187 51.94797611093753, -114.4742908039062 51.94055257031255))',4269).STStartPoint()).STAsText(),4269)",), ("GEOGRAPHY::STGeomFromText(GEOMETRY::STGeomFromText('POLYGON ((-112.1157004828125 49.53477394218754, -112.1156657703125 49.52339953203125, -112.124386 ... (704 characters truncated) ... 1212683476562 49.53115973984376, -112.1212804195312 49.53477826953127, -112.1157004828125 49.53477394218754))',4269).STStartPoint()).STAsText(),4269)",))]
(Background on this error at: http://sqlalche.me/e/9h9h)
我假设错误是由于周围的引号导致了一个字符串，而不是传递给 SQL 的函数。
我尝试在 bind_processor return 字符串上使用 sqlalchemy.sql.expression.text 但我收到此错误：
ProgrammingError: (pyodbc.ProgrammingError) ('Invalid parameter type. param-index=0 param-type=TextClause', 'HY105')
[SQL: INSERT INTO [Spatial_Table] (geometry) VALUES (?)]
[parameters: (<sqlalchemy.sql.elements.TextClause object at 0x000002566D5A40F0>,)]
(Background on this error at: http://sqlalche.me/e/f405)
而且我看过sqlalchemy.sql.expression.func；但是，我不知道如何将它与复杂的 GEOGRAPHY::STGeomFromText 方法一起使用。
虽然我的推测可能不正确，因为这将适用于 SQL:
INSERT INTO dbo.[Spatial_Table] ([geometry])
VALUES ('POLYGON((-110.17315242968752 52.66767554218751,-110.18536282187502 52.66770066015627,-110.19718901640618 52.66771763203127,-110.197593865625 52.667718411718795,-110.19747227656248 52.67594785000003,-110.1732282007812 52.67592660234379,-110.17315242968752 52.66767554218751))')
只需使用：
class Geography(sqlalchemy.types.UserDefinedType):
def get_col_spec(self):
return "GEOGRAPHY"
将导致此错误，即使参数看起来是正确的。它仍然出现相同的操作数类型冲突错误：
DataError: (pyodbc.DataError) ('22018', '[22018] [Microsoft][ODBC Driver 13 for SQL Server][SQL Server]Operand type clash: ntext is incompatible with geography (206) (SQLExecDirectW); [22018] [Microsoft][ODBC Driver 13 for SQL Server][SQL Server]Statement(s) could not be prepared. (8180)')
[SQL: INSERT INTO [Spatial_Table] (geometry) VALUES (?)]
[parameters: ('POLYGON ((-114.4742908039062 51.94055257031255, -114.4623163671875 51.94054674921875, -114.4506284421875 51.94053819687502, -114.4506230351562 51.933 ... (1739 characters truncated) ... -114.49781949375 51.93693783750001, -114.4978060703125 51.9479693648438, -114.4743004117187 51.94797611093753, -114.4742908039062 51.94055257031255))',)]
(Background on this error at: http://sqlalche.me/e/9h9h)
这是一个完整的例子：
import geopandas
import pyodbc
import urllib
import sqlalchemy
params = 'DRIVER={ODBC Driver 13 for SQL Server};' \
'SERVER=ServerName;' \
'PORT=1433;' \
'DATABASE=DatabaseName;' \
'trusted_connection=yes;'
params = urllib.parse.quote_plus(params)
db = sqlalchemy.create_engine('mssql+pyodbc:///?odbc_connect=%s' % params)
class Geography(sqlalchemy.types.UserDefinedType):
def get_col_spec(self):
return "GEOGRAPHY"
def bind_processor(self, dialect):
def process(value):
if value is None:
return None
return 'GEOGRAPHY::STGeomFromText(GEOMETRY::STGeomFromText({0},4269).MakeValid().STUnion(GEOMETRY::STGeomFromText({0},4269).STStartPoint()).STAsText(),4269)'.format("'" + value + "'")
return process
gdf = geopandas.GeoDataFrame({'geometry':['POLYGON ((-114.4742908039062 51.94055257031255, -114.4623163671875 51.94054674921875, -114.4506284421875 51.94053819687502, -114.4506230351562 51.93329010000002, -114.4506172351562 51.92605222890626, -114.4270012117187 51.92605734296876, -114.42699041875 51.91881815312502, -114.4269793164062 51.91139827421875, -114.4150635234375 51.91142824140627, -114.4150633671875 51.90417951171878, -114.41506316875 51.89693553984375, -114.403441165625 51.89697914921879, -114.3914711632812 51.89696492578128, -114.3914339117187 51.88973628046875, -114.3913953804687 51.88226671015627, -114.3677831210937 51.88229111953126, -114.3677989414062 51.87505186875001, -114.3678146921875 51.86781865390628, -114.3678270578125 51.86058223671876, -114.3678396773437 51.8531650226563, -114.3561627734375 51.85316232265626, -114.34419459375 51.85315451953124, -114.34420228125 51.8459314015625, -114.3442099304687 51.83869999843751, -114.3561678164062 51.83869954843755, -114.3561729164063 51.83145872656252, -114.3561489476562 51.82404763359375, -114.3678250734375 51.82404504843754, -114.367827496875 51.82784562734378, -114.3678343414063 51.83869510312502, -114.3797993828125 51.83869063437504, -114.3797996945312 51.84592788906252, -114.3798041078125 51.85316721640629, -114.3914338414062 51.85316111093755, -114.3914139953125 51.86780720156253, -114.4033583140625 51.86778718046878, -114.4150253679687 51.86777429218751, -114.415020278125 51.8822367257813, -114.432806471875 51.88222315703126, -114.4505780710937 51.882230434375, -114.450589053125 51.88963689218753, -114.450594271875 51.89689434218752, -114.4622547703125 51.89689832265628, -114.4622642132812 51.90414635312504, -114.4622736234375 51.91139415781254, -114.4742338164062 51.91140009453125, -114.4742472351562 51.91883031875, -114.4742673734375 51.92607473984378, -114.4858996257812 51.92606931250003, -114.4978195835938 51.92606766953128, -114.49781949375 51.93693783750001, -114.4978060703125 51.9479693648438, -114.4743004117187 51.94797611093753, -114.4742908039062 51.94055257031255))']})
gdf.to_sql('Spatial_Table',
if_exists='replace',
index=False,
dtype={'geometry': Geography},
con=db)
我希望在 SQLAlchemy 或 GeoAlchemy 中看到直接 support for Geography and Geometry for MS SQL Server。
而不是在 Python 中进行值处理的 bind_processor()，使用 bind_expression() 进行 SQL 端处理：
class Geography(sqlalchemy.types.UserDefinedType):
def get_col_spec(self):
return "GEOGRAPHY"
def bind_expression(self, bindvalue):
# Note that this does *not* format the value to the expression text, but
# the bind value key.
return text(f'GEOGRAPHY::STGeomFromText(GEOMETRY::STGeomFromText(:{bindvalue.key},4269).MakeValid().STUnion(GEOMETRY::STGeomFromText(:{bindvalue.key},4269).STStartPoint()).STAsText(),4269)').bindparams(bindvalue)
我刚遇到同样的问题并找到了解决方案。这可能有帮助：
from sqlalchemy import func
from sqlalchemy.sql.expression import text
from sqlalchemy.sql.schema import Column
from sqlalchemy.types import UserDefinedType, TypeEngine
from geojson_funcs import wkt2geojson, geojson2wkt
class Geography(UserDefinedType):
cache_ok = True
def __init__(self, srid: int = 4326):
self.srid = srid
def get_col_spec(self):
return "GEOGRAPHY"
def bind_expression(self, bindvalue):
# not able to use func here since "::" in function name
exp = f"geography::STGeomFromText(:{bindvalue.key}, {self.srid})"
f = text(exp)
return f.bindparams(bindvalue)
def column_expression(self, col: Column):
# not able to use func here since function needs to be applied as method to the column itself
# col.func() not func(col)
exp = f"{col.key}.STAsText()"
f1 = text(exp)
f2 = func.IIF(col == None, None, f1, type_=self)
return f2
def bind_processor(self, dialect):
def process(value):
return geojson2wkt(value) if value else value
return process
def result_processor(self, dialect, coltype):
def process(value):
return wkt2geojson(value) if value else value
return process
class comparator_factory(TypeEngine.Comparator):
def __add__(self, other):
return self.op("goofy")(other)
def intersects_bounds(self, lon_min, lon_max, lat_min, lat_max):
p = f'POLYGON (({lon_min} {lat_min}, {lon_max} {lat_min}, {lon_max} {lat_max}, {lon_min} {lat_max}, {lon_min} {lat_min}))'
exp = f"geography::STGeomFromText('{p}', {self.type.srid}).STIntersects({self.expr.name}) = 1"
f = text(exp)
return f
我唯一的问题是代码中的注释：
- bind_expression: 由于函数名称中的“::” ，无法在此处使用 func
- column_expression：无法在此处使用 func，因为需要将函数作为方法应用于列本身 -> col.func() not func (col)
如果有人知道如何解决这些问题，我很乐意在这里更新我的示例:)
此外，我将此函数用于 return GeoJson 而不是 WKT：
from typing import Union
import geojson
import shapely.wkt
from shapely.geometry import shape, GeometryCollection, Point, MultiPoint, \
LineString, MultiLineString, Polygon, MultiPolygon
GeoJsonType: type = Union[geojson.Feature, geojson.FeatureCollection]
WktType: type = Union[GeometryCollection, Point, MultiPoint, LineString, MultiLineString, Polygon, MultiPolygon]
def wkt2geojson(wkt: str) -> GeoJsonType:
wkt = shapely.wkt.loads(wkt)
wkt = validate_wkt(wkt)
if type(wkt) == GeometryCollection:
features = [geojson.Feature(geometry=o, properties={}) for o in wkt.geoms]
feature_collection = geojson.FeatureCollection(features)
return feature_collection
else:
feature = geojson.Feature(geometry=wkt, properties={})
return feature
def geojson2wkt(data: Union[GeoJsonType, dict]) -> WktType:
data = validate_geojson(data)
if data.get('type') == 'Feature':
g = shape(data.get('geometry'))
return g.wkt
elif data.get('type') == 'FeatureCollection':
g = [shape(f.get('geometry')) for f in data.get('features')]
gc = GeometryCollection(g)
return gc.wkt
else:
raise Exception
def validate_geojson(data: Union[GeoJsonType, dict]) -> GeoJsonType:
data = geojson.loads(geojson.dumps(data))
if not data.is_valid:
raise ValueError('value is not a valid GeoJson')
return data
def validate_wkt(data: WktType) -> WktType:
if not data.is_valid:
raise ValueError('value is not a valid WKT')
return data