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.ext.declarative 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
44 metadata = MetaData()
45 Base = declarative_base(metadata=metadata)
46
47
48 def _compile_buffer_default(element, compiler, **kw):
49 """Compile the element in the default case (no specific dialect).
50
51 This function should not be needed for SQLAlchemy >= 1.1.
52 """
53 return '{}({})'.format('ST_Buffer', compiler.process(element.clauses, **kw))
54
55
56 def _compile_buffer_sqlite(element, compiler, **kw):
57 """Compile the element for the SQLite dialect."""
58 # Get the side parameters
59 compiled = compiler.process(element.clauses, **kw)
60 side_params = [
61 i for i in element.clauses
62 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)
88 compiles(functions.ST_Buffer, 'sqlite')(_compile_buffer_sqlite)
89
90
91 def test_specific_compilation(conn):
92 # Build a query with a sided buffer
93 query = select([
94 func.ST_AsText(
95 func.ST_Buffer(WKTElement('LINESTRING(0 0, 1 0)', srid=4326), 1, 'side=left')
96 )
97 ])
98
99 # Check the compiled query: the sided buffer should appear only in the SQLite query
100 compiled_query = str(query.compile(dialect=conn.dialect))
101 if conn.dialect.name == 'sqlite':
102 assert 'SingleSidedBuffer' in compiled_query
103 assert 'ST_Buffer' not in compiled_query
104 else:
105 assert 'SingleSidedBuffer' not in compiled_query
106 assert 'ST_Buffer' in compiled_query
107
108 # Check the actual result of the query
109 res = conn.execute(query).scalar()
110 assert format_wkt(res) == 'POLYGON((1 0,0 0,0 1,1 1,1 0))'
111
112 # Build a query with symmetric buffer to check nothing was broken
113 query = select([
114 func.ST_AsText(
115 func.ST_Buffer(WKTElement('LINESTRING(0 0, 1 0)', srid=4326), 1)
116 )
117 ])
118
119 # Check the compiled query: the sided buffer should never appear in the query
120 compiled_query = str(query.compile(dialect=conn.dialect))
121 assert 'SingleSidedBuffer' not in compiled_query
122 if conn.dialect.name == 'sqlite':
123 assert 'ST_Buffer' not in compiled_query
124 assert 'Buffer' in compiled_query
125 else:
126 assert 'ST_Buffer' in compiled_query
127
128 # Check the actual result of the query
129 res = conn.execute(query).scalar()
130 assert format_wkt(res) != 'POLYGON((1 0,0 0,0 1,1 1,1 0))'
131 assert format_wkt(res).startswith('POLYGON((1 1,1')