expression
is a callable within the sqlalchemy.sql
module of the SQLAlchemy project.
ClauseElement,
Select,
column,
extract,
functions,
operators,
schema,
select,
sqltypes,
and table
are several other callables with code examples from the same sqlalchemy.sql
package.
GeoAlchemy2 (project documentation and PyPI package information) extends SQLAlchemy with new data types for working with geospatial databases, particularly PostGIS, which is a spatial database extender for PostgreSQL. The project is provided as open source under the MIT license.
GeoAlchemy2 / geoalchemy2 / init.py
# __init__.py
from .types import ( # NOQA
Geometry,
Geography,
Raster
)
from .elements import ( # NOQA
WKTElement,
WKBElement,
RasterElement
)
from .exc import ArgumentError
from . import functions # NOQA
from . import types # NOQA
from sqlalchemy import Table, event
from sqlalchemy.sql import select, func, expression
def _setup_ddl_event_listeners():
@event.listens_for(Table, "before_create")
def before_create(target, connection, **kw):
dispatch("before-create", target, connection)
@event.listens_for(Table, "after_create")
def after_create(target, connection, **kw):
dispatch("after-create", target, connection)
@event.listens_for(Table, "before_drop")
def before_drop(target, connection, **kw):
dispatch("before-drop", target, connection)
@event.listens_for(Table, "after_drop")
def after_drop(target, connection, **kw):
dispatch("after-drop", target, connection)
def dispatch(event, table, bind):
if event in ('before-create', 'before-drop'):
gis_cols = [c for c in table.c if
isinstance(c.type, Geometry) and
c.type.management is True]
regular_cols = [x for x in table.c if x not in gis_cols]
table.info["_saved_columns"] = table.c
column_collection = expression.ColumnCollection()
for col in regular_cols:
column_collection.add(col)
table.columns = column_collection
if event == 'before-drop':
for c in gis_cols:
if bind.dialect.name == 'sqlite':
drop_func = 'DiscardGeometryColumn'
elif bind.dialect.name == 'postgresql':
drop_func = 'DropGeometryColumn'
else:
raise ArgumentError('dialect {} is not supported'.format(bind.dialect.name))
args = [table.schema] if table.schema else []
args.extend([table.name, c.name])
stmt = select([getattr(func, drop_func)(*args)])
stmt = stmt.execution_options(autocommit=True)
bind.execute(stmt)
elif event == 'after-create':
table.columns = table.info.pop('_saved_columns')
for c in table.c:
if isinstance(c.type, Geometry) and c.type.management is True:
## ... source file continues with no further expression examples...