tessl/pypi-geoalchemy2

Using SQLAlchemy with Spatial Databases

—

Pending

Quality

Pending

Does it follow best practices?

Impact

Pending

No eval scenarios have been run

Overview

Eval results

Files

Migration Support

Name: tessl/pypi-geoalchemy2
Author: tessl

Comprehensive Alembic integration for managing spatial database schema migrations, including specialized operations for spatial columns, tables, and indexes with support for all major spatial database backends.

Installation and Setup

Migration support requires Alembic to be installed:

pip install alembic

Configure Alembic environment for spatial migrations:

# alembic/env.py
from geoalchemy2.alembic_helpers import writer
from geoalchemy2 import Geometry, Geography, Raster

# Register GeoAlchemy2 types for autogeneration
target_metadata = Base.metadata

# Configure rewriter for spatial operations
context.configure(
    # ... other configuration
    render_item=writer.render_item,
    compare_type=True,
    compare_server_default=True
)

Capabilities

Spatial Column Operations

Specialized operations for adding and removing spatial columns with proper metadata and index management.

class AddGeospatialColumnOp:
    """
    Operation for adding spatial columns to existing tables.
    
    Handles:
    - Spatial column creation with proper constraints
    - Automatic spatial index generation
    - Spatial metadata registration
    - Database-specific implementation details
    """

class DropGeospatialColumnOp:
    """
    Operation for removing spatial columns from tables.
    
    Handles:
    - Spatial index removal
    - Spatial metadata cleanup  
    - Constraint removal
    - Cross-database compatibility
    """

Usage examples:

# In Alembic migration file
from alembic import op
from geoalchemy2 import Geometry
from geoalchemy2.alembic_helpers import AddGeospatialColumnOp, DropGeospatialColumnOp

def upgrade():
    # Add spatial column
    op.invoke(AddGeospatialColumnOp(
        table_name='buildings',
        column=Column('geom', Geometry('POLYGON', srid=4326))
    ))
    
    # Standard Alembic syntax also works
    op.add_column('buildings', 
        Column('location', Geometry('POINT', srid=4326))
    )

def downgrade():
    # Remove spatial column
    op.invoke(DropGeospatialColumnOp(
        table_name='buildings',
        column_name='geom'
    ))
    
    op.drop_column('buildings', 'location')

Spatial Table Operations

Operations for creating and dropping entire spatial tables with proper setup.

class CreateGeospatialTableOp:
    """
    Operation for creating tables with spatial columns.
    
    Handles:
    - Table creation with spatial column definitions
    - Automatic spatial index creation
    - Spatial metadata initialization
    - Database-specific optimizations
    """

class DropGeospatialTableOp:
    """
    Operation for dropping tables containing spatial columns.
    
    Handles:
    - Spatial metadata cleanup
    - Spatial index removal
    - Proper table dropping order
    - Cross-database cleanup
    """

Usage examples:

def upgrade():
    # Create spatial table
    op.invoke(CreateGeospatialTableOp(
        table_name='parcels',
        columns=[
            Column('id', Integer, primary_key=True),
            Column('parcel_id', String(50)),
            Column('geom', Geometry('POLYGON', srid=4326)),
            Column('centroid', Geometry('POINT', srid=4326))
        ]
    ))

def downgrade():
    # Drop spatial table
    op.invoke(DropGeospatialTableOp(table_name='parcels'))

Spatial Index Operations

Dedicated operations for managing spatial indexes independently of columns and tables.

class CreateGeospatialIndexOp:
    """
    Operation for creating spatial indexes.
    
    Handles:
    - Database-specific spatial index creation
    - Index naming conventions
    - Advanced index options (N-D, partial indexes)
    - Performance optimization settings
    """

class DropGeospatialIndexOp:
    """
    Operation for dropping spatial indexes.
    
    Handles:
    - Safe index removal
    - Database-specific cleanup
    - Index dependency management
    """

Usage examples:

def upgrade():
    # Create spatial index
    op.invoke(CreateGeospatialIndexOp(
        index_name='idx_buildings_geom_spatial',
        table_name='buildings',
        column_name='geom',
        index_type='gist',  # PostgreSQL
        postgresql_ops={'geom': 'gist_geometry_ops_nd'}  # N-D index
    ))

def downgrade():
    # Drop spatial index
    op.invoke(DropGeospatialIndexOp(
        index_name='idx_buildings_geom_spatial'
    ))

Database Implementation Support

Specialized implementations for different database backends with appropriate spatial handling.

class GeoPackageImpl:
    """
    Alembic implementation for GeoPackage databases.
    
    Extends SQLiteImpl with GeoPackage-specific handling:
    - OGC GeoPackage standard compliance
    - Spatial reference system management
    - Built-in R-tree spatial indexing
    """

Autogeneration Support

Automatic detection and generation of spatial schema changes:

Type Comparison

# alembic/env.py configuration for spatial type detection
def compare_type(context, inspected_column, metadata_column,
                inspected_type, metadata_type):
    # Handle spatial type changes
    if isinstance(metadata_type, (Geometry, Geography, Raster)):
        return not spatial_types_equal(inspected_type, metadata_type)
    return None

def spatial_types_equal(inspected, metadata):
    """Compare spatial types for schema changes."""
    if type(inspected) != type(metadata):
        return False
    
    # Compare geometry_type, srid, dimension
    return (
        getattr(inspected, 'geometry_type', None) == 
        getattr(metadata, 'geometry_type', None) and
        getattr(inspected, 'srid', -1) == 
        getattr(metadata, 'srid', -1)
    )

Automatic Migration Generation

# Generate migration with spatial changes
$ alembic revision --autogenerate -m "Add spatial columns"

# Generated migration includes spatial operations
def upgrade():
    # ### commands auto generated by Alembic ###
    op.add_column('locations', 
        sa.Column('geom', geoalchemy2.types.Geometry(
            geometry_type='POINT', srid=4326), nullable=True))
    op.create_geospatial_index('idx_locations_geom_spatial', 
        'locations', ['geom'])
    # ### end Alembic commands ###

Migration Workflows

Adding Spatial Capabilities to Existing Database

Migrate non-spatial database to support spatial operations:

# Migration: Add spatial extension
def upgrade():
    # PostgreSQL: Enable PostGIS
    op.execute('CREATE EXTENSION IF NOT EXISTS postgis')
    
    # Add spatial columns to existing tables
    op.add_column('addresses', 
        Column('location', Geometry('POINT', srid=4326))
    )
    
    # Create spatial indexes
    op.create_index('idx_addresses_location_spatial',
        'addresses', ['location'], postgresql_using='gist')

def downgrade():
    op.drop_index('idx_addresses_location_spatial')
    op.drop_column('addresses', 'location')
    # Note: PostGIS extension cleanup may require manual intervention

Changing Spatial Reference Systems

Migrate spatial data between coordinate systems:

def upgrade():
    # Add new column with different SRID
    op.add_column('parcels',
        Column('geom_utm', Geometry('POLYGON', srid=3857))
    )
    
    # Transform existing data
    op.execute('''
        UPDATE parcels 
        SET geom_utm = ST_Transform(geom, 3857)
        WHERE geom IS NOT NULL
    ''')
    
    # Drop old column (in separate migration for safety)
    
def downgrade():
    op.drop_column('parcels', 'geom_utm')

Cross-Database Migration

Migrate spatial data between different database backends:

# From PostgreSQL to SQLite with SpatiaLite
def upgrade():
    # Initialize SpatiaLite (handled automatically)
    pass

def downgrade():
    pass

# Data migration script (separate from schema migration)
def migrate_spatial_data():
    # Source: PostgreSQL
    source_engine = create_engine('postgresql://...')
    
    # Target: SQLite with SpatiaLite  
    target_engine = create_engine('sqlite:///spatial.db')
    
    # Read from source
    source_data = source_engine.execute(
        'SELECT id, name, ST_AsEWKT(geom) as geom_wkt FROM spatial_table'
    ).fetchall()
    
    # Write to target
    for row in source_data:
        target_engine.execute(
            'INSERT INTO spatial_table (id, name, geom) VALUES (?, ?, GeomFromEWKT(?))',
            (row.id, row.name, row.geom_wkt)
        )

Schema Evolution Patterns

Common patterns for evolving spatial schemas:

# Pattern 1: Add spatial constraint
def upgrade():
    # Add constraint to existing geometry column
    op.execute('''
        ALTER TABLE buildings 
        ADD CONSTRAINT enforce_geom_type 
        CHECK (geometrytype(geom) = 'POLYGON' OR geom IS NULL)
    ''')

# Pattern 2: Partition spatial data
def upgrade():
    # Create partitioned table for large spatial datasets
    op.execute('''
        CREATE TABLE spatial_data_2024 (
            LIKE spatial_data INCLUDING ALL
        ) INHERITS (spatial_data)
    ''')
    
    # Add partition constraint
    op.execute('''
        ALTER TABLE spatial_data_2024
        ADD CONSTRAINT year_2024 
        CHECK (extract(year from created_at) = 2024)
    ''')

# Pattern 3: Upgrade geometry dimension
def upgrade():
    # Convert 2D geometries to 3D
    op.execute('''
        UPDATE elevation_points 
        SET geom = ST_Force3D(geom)
        WHERE ST_CoordDim(geom) = 2
    ''')

Error Handling and Rollback

Robust error handling for spatial migrations:

def upgrade():
    try:
        # Spatial operations in transaction
        with op.get_context().autocommit_block():
            op.add_column('features', 
                Column('geom', Geometry('POLYGON', srid=4326))
            )
            
            # Validate spatial data
            result = op.get_bind().execute('''
                SELECT COUNT(*) FROM features 
                WHERE geom IS NOT NULL AND NOT ST_IsValid(geom)
            ''').scalar()
            
            if result > 0:
                raise ValueError(f"Found {result} invalid geometries")
                
    except Exception as e:
        # Rollback spatial changes
        op.drop_column('features', 'geom')
        raise

def downgrade():
    # Safe downgrade with data preservation
    invalid_geoms = op.get_bind().execute('''
        SELECT id FROM features WHERE NOT ST_IsValid(geom)
    ''').fetchall()
    
    if invalid_geoms:
        print(f"Warning: {len(invalid_geoms)} invalid geometries will be lost")
    
    op.drop_column('features', 'geom')

Performance Considerations

Optimize spatial migrations for large datasets:

Batch Processing

def upgrade():
    # Process large spatial updates in batches
    batch_size = 10000
    offset = 0
    
    while True:
        result = op.get_bind().execute(f'''
            UPDATE large_spatial_table 
            SET geom_transformed = ST_Transform(geom, 3857)
            WHERE id BETWEEN {offset} AND {offset + batch_size - 1}
            AND geom_transformed IS NULL
        ''')
        
        if result.rowcount == 0:
            break
            
        offset += batch_size
        print(f"Processed {offset} rows")

Index Management

def upgrade():
    # Drop spatial indexes before bulk operations
    op.drop_index('idx_features_geom_spatial')
    
    # Perform bulk spatial updates
    op.execute('UPDATE features SET geom = ST_Transform(geom, 3857)')
    
    # Recreate spatial indexes
    op.create_index('idx_features_geom_spatial', 'features', ['geom'],
                   postgresql_using='gist')

Install with Tessl CLI