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:

  1. 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)
  1. 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 if isinstance(i, BindParameter) and "side" in str(i.value)
 62     ]
 63
 64     if side_params:
 65         side_param = side_params[0]
 66         if "right" in side_param.value:
 67             # If the given side is 'right', we translate the value into 0 and switch to the sided
 68             # function
 69             side_param.value = 0
 70             element.identifier = "SingleSidedBuffer"
 71         elif "left" in side_param.value:
 72             # If the given side is 'left', we translate the value into 1 and switch to the sided
 73             # function
 74             side_param.value = 1
 75             element.identifier = "SingleSidedBuffer"
 76
 77     if element.identifier == "ST_Buffer":
 78         # If the identifier is still the default ST_Buffer we switch to the SpatiaLite function
 79         element.identifier = "Buffer"
 80
 81     # If there is no side parameter or if the side value is 'both', we use the default function
 82     return "{}({})".format(element.identifier, compiled)
 83
 84
 85 # Register the specific compilation rules
 86 compiles(functions.ST_Buffer)(_compile_buffer_default)
 87 compiles(functions.ST_Buffer, "sqlite")(_compile_buffer_sqlite)
 88
 89
 90 def test_specific_compilation(conn):
 91     # Build a query with a sided buffer
 92     query = select(
 93         [
 94             func.ST_AsText(
 95                 func.ST_Buffer(WKTElement("LINESTRING(0 0, 1 0)", srid=4326), 1, "side=left")
 96             )
 97         ]
 98     )
 99
100     # Check the compiled query: the sided buffer should appear only in the SQLite query
101     compiled_query = str(query.compile(dialect=conn.dialect))
102     if conn.dialect.name == "sqlite":
103         assert "SingleSidedBuffer" in compiled_query
104         assert "ST_Buffer" not in compiled_query
105     else:
106         assert "SingleSidedBuffer" not in compiled_query
107         assert "ST_Buffer" in compiled_query
108
109     # Check the actual result of the query
110     res = conn.execute(query).scalar()
111     assert format_wkt(res) == "POLYGON((1 0,0 0,0 1,1 1,1 0))"
112
113     # Build a query with symmetric buffer to check nothing was broken
114     query = select(
115         [func.ST_AsText(func.ST_Buffer(WKTElement("LINESTRING(0 0, 1 0)", srid=4326), 1))]
116     )
117
118     # Check the compiled query: the sided buffer should never appear in the query
119     compiled_query = str(query.compile(dialect=conn.dialect))
120     assert "SingleSidedBuffer" not in compiled_query
121     if conn.dialect.name == "sqlite":
122         assert "ST_Buffer" not in compiled_query
123         assert "Buffer" in compiled_query
124     else:
125         assert "ST_Buffer" in compiled_query
126
127     # Check the actual result of the query
128     res = conn.execute(query).scalar()
129     assert format_wkt(res) != "POLYGON((1 0,0 0,0 1,1 1,1 0))"
130     assert format_wkt(res).startswith("POLYGON((1 1,1")

Gallery generated by Sphinx-Gallery