Note
Go to the end to download the full example code.
Function translation for specific dialectΒΆ
Some functions have different names depending on the dialect. But sometimes one function in one dialect can be mapped to several other functions in another dialect, depending on the arguments passed. For example, the ST_Buffer function in PostgreSQL can translate into 2 functions in SQLite:
if the buffer is two-sided (symmetric), the PostgreSQL function:
ST_Buffer(the_table.geom, 10)
should become in SQLite:
Buffer(the_table.geom, 10)
if the buffer is one-sided, the PostgreSQL function:
ST_Buffer(the_table.geom, 10, 'side=right')
should become in SQLite:
SingleSidedBuffer(the_table.geom, 10, 0)
This case is much more complicated than just mapping a function name and we show here how to deal with it.
This example uses SQLAlchemy core queries.
32 from sqlalchemy import MetaData
33 from sqlalchemy import func
34 from sqlalchemy.ext.compiler import compiles
35 from sqlalchemy.orm import declarative_base
36 from sqlalchemy.sql.expression import BindParameter
37
38 from geoalchemy2 import WKTElement
39 from geoalchemy2 import functions
40
41 # Tests imports
42 from tests import format_wkt
43 from tests import select
44 from tests import test_only_with_dialects
45
46 metadata = MetaData()
47 Base = declarative_base(metadata=metadata)
48
49
50 def _compile_buffer_default(element, compiler, **kw):
51 """Compile the element in the default case (no specific dialect).
52
53 This function should not be needed for SQLAlchemy >= 1.1.
54 """
55 return "{}({})".format("ST_Buffer", compiler.process(element.clauses, **kw))
56
57
58 def _compile_buffer_sqlite(element, compiler, **kw):
59 """Compile the element for the SQLite dialect."""
60 # Get the side parameters
61 compiled = compiler.process(element.clauses, **kw)
62 side_params = [
63 i for i in element.clauses if isinstance(i, BindParameter) and "side" in str(i.value)
64 ]
65
66 if side_params:
67 side_param = side_params[0]
68 if "right" in side_param.value:
69 # If the given side is 'right', we translate the value into 0 and switch to the sided
70 # function
71 side_param.value = 0
72 element.identifier = "SingleSidedBuffer"
73 elif "left" in side_param.value:
74 # If the given side is 'left', we translate the value into 1 and switch to the sided
75 # function
76 side_param.value = 1
77 element.identifier = "SingleSidedBuffer"
78
79 if element.identifier == "ST_Buffer":
80 # If the identifier is still the default ST_Buffer we switch to the SpatiaLite function
81 element.identifier = "Buffer"
82
83 # If there is no side parameter or if the side value is 'both', we use the default function
84 return "{}({})".format(element.identifier, compiled)
85
86
87 # Register the specific compilation rules
88 compiles(functions.ST_Buffer)(_compile_buffer_default) # type: ignore
89 compiles(functions.ST_Buffer, "sqlite")(_compile_buffer_sqlite) # type: ignore
90 compiles(functions.ST_Buffer, "geopackage")(_compile_buffer_sqlite) # type: ignore
91
92
93 @test_only_with_dialects("postgresql", "sqlite")
94 def test_specific_compilation(conn):
95 # Build a query with a sided buffer
96 query = select(
97 [
98 func.ST_AsText(
99 func.ST_Buffer(WKTElement("LINESTRING(0 0, 1 0)", srid=4326), 1, "side=left")
100 )
101 ]
102 )
103
104 # Check the compiled query: the sided buffer should appear only in the SQLite query
105 compiled_query = str(query.compile(dialect=conn.dialect))
106 if conn.dialect.name in ["sqlite", "geopackage"]:
107 assert "SingleSidedBuffer" in compiled_query
108 assert "ST_Buffer" not in compiled_query
109 else:
110 assert "SingleSidedBuffer" not in compiled_query
111 assert "ST_Buffer" in compiled_query
112
113 # Check the actual result of the query
114 res = conn.execute(query).scalar()
115 assert format_wkt(res) == "POLYGON((1 0,0 0,0 1,1 1,1 0))"
116
117 # Build a query with symmetric buffer to check nothing was broken
118 query = select(
119 [func.ST_AsText(func.ST_Buffer(WKTElement("LINESTRING(0 0, 1 0)", srid=4326), 1))]
120 )
121
122 # Check the compiled query: the sided buffer should never appear in the query
123 compiled_query = str(query.compile(dialect=conn.dialect))
124 assert "SingleSidedBuffer" not in compiled_query
125 if conn.dialect.name in ["sqlite", "geopackage"]:
126 assert "ST_Buffer" not in compiled_query
127 assert "Buffer" in compiled_query
128 else:
129 assert "ST_Buffer" in compiled_query
130
131 # Check the actual result of the query
132 res = conn.execute(query).scalar()
133 assert format_wkt(res) != "POLYGON((1 0,0 0,0 1,1 1,1 0))"
134 assert format_wkt(res).startswith("POLYGON((1 1,1")