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