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:

  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.

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

Gallery generated by Sphinx-Gallery