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.

 31 from sqlalchemy import MetaData
 32 from sqlalchemy import func
 33 from sqlalchemy import select
 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 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 f"{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         func.ST_AsText(
 97             func.ST_Buffer(WKTElement("LINESTRING(0 0, 1 0)", srid=4326), 1, "side=left")
 98         )
 99     )
100
101     # Check the compiled query: the sided buffer should appear only in the SQLite query
102     compiled_query = str(query.compile(dialect=conn.dialect))
103     if conn.dialect.name in ["sqlite", "geopackage"]:
104         assert "SingleSidedBuffer" in compiled_query
105         assert "ST_Buffer" not in compiled_query
106     else:
107         assert "SingleSidedBuffer" not in compiled_query
108         assert "ST_Buffer" in compiled_query
109
110     # Check the actual result of the query
111     res = conn.execute(query).scalar()
112     assert format_wkt(res) == "POLYGON((1 0,0 0,0 1,1 1,1 0))"
113
114     # Build a query with symmetric buffer to check nothing was broken
115     query = select(func.ST_AsText(func.ST_Buffer(WKTElement("LINESTRING(0 0, 1 0)", srid=4326), 1)))
116
117     # Check the compiled query: the sided buffer should never appear in the query
118     compiled_query = str(query.compile(dialect=conn.dialect))
119     assert "SingleSidedBuffer" not in compiled_query
120     if conn.dialect.name in ["sqlite", "geopackage"]:
121         assert "ST_Buffer" not in compiled_query
122         assert "Buffer" in compiled_query
123     else:
124         assert "ST_Buffer" in compiled_query
125
126     # Check the actual result of the query
127     res = conn.execute(query).scalar()
128     assert format_wkt(res) != "POLYGON((1 0,0 0,0 1,1 1,1 0))"
129     assert format_wkt(res).startswith("POLYGON((1 1,1")

Gallery generated by Sphinx-Gallery