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)

2. 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 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         [
     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