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.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")

Gallery generated by Sphinx-Gallery