Note
Click here 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:
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)
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 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 | from sqlalchemy import MetaData from sqlalchemy import func from sqlalchemy.ext.compiler import compiles from sqlalchemy.ext.declarative import declarative_base from sqlalchemy.sql.expression import BindParameter from geoalchemy2 import WKTElement from geoalchemy2 import functions # Tests imports from tests import format_wkt from tests import select metadata = MetaData() Base = declarative_base(metadata=metadata) def _compile_buffer_default(element, compiler, **kw): """Compile the element in the default case (no specific dialect). This function should not be needed for SQLAlchemy >= 1.1. """ return '{}({})'.format('ST_Buffer', compiler.process(element.clauses, **kw)) def _compile_buffer_sqlite(element, compiler, **kw): """Compile the element for the SQLite dialect.""" # Get the side parameters compiled = compiler.process(element.clauses, **kw) side_params = [ i for i in element.clauses if isinstance(i, BindParameter) and 'side' in str(i.value) ] if side_params: side_param = side_params[0] if 'right' in side_param.value: # If the given side is 'right', we translate the value into 0 and switch to the sided # function side_param.value = 0 element.identifier = 'SingleSidedBuffer' elif 'left' in side_param.value: # If the given side is 'left', we translate the value into 1 and switch to the sided # function side_param.value = 1 element.identifier = 'SingleSidedBuffer' if element.identifier == 'ST_Buffer': # If the identifier is still the default ST_Buffer we switch to the SpatiaLite function element.identifier = 'Buffer' # If there is no side parameter or if the side value is 'both', we use the default function return '{}({})'.format(element.identifier, compiled) # Register the specific compilation rules compiles(functions.ST_Buffer)(_compile_buffer_default) compiles(functions.ST_Buffer, 'sqlite')(_compile_buffer_sqlite) def test_specific_compilation(conn): # Build a query with a sided buffer query = select([ func.ST_AsText( func.ST_Buffer(WKTElement('LINESTRING(0 0, 1 0)', srid=4326), 1, 'side=left') ) ]) # Check the compiled query: the sided buffer should appear only in the SQLite query compiled_query = str(query.compile(dialect=conn.dialect)) if conn.dialect.name == 'sqlite': assert 'SingleSidedBuffer' in compiled_query assert 'ST_Buffer' not in compiled_query else: assert 'SingleSidedBuffer' not in compiled_query assert 'ST_Buffer' in compiled_query # Check the actual result of the query res = conn.execute(query).scalar() assert format_wkt(res) == 'POLYGON((1 0,0 0,0 1,1 1,1 0))' # Build a query with symmetric buffer to check nothing was broken query = select([ func.ST_AsText( func.ST_Buffer(WKTElement('LINESTRING(0 0, 1 0)', srid=4326), 1) ) ]) # Check the compiled query: the sided buffer should never appear in the query compiled_query = str(query.compile(dialect=conn.dialect)) assert 'SingleSidedBuffer' not in compiled_query if conn.dialect.name == 'sqlite': assert 'ST_Buffer' not in compiled_query assert 'Buffer' in compiled_query else: assert 'ST_Buffer' in compiled_query # Check the actual result of the query res = conn.execute(query).scalar() assert format_wkt(res) != 'POLYGON((1 0,0 0,0 1,1 1,1 0))' assert format_wkt(res).startswith('POLYGON((1 1,1') |