sqlalchemy-spanner

import os from sqlalchemy import create_engine, text, MetaData, Table, Column, String, Integer # Set these environment variables or replace directly project_id = os.environ.get('SPANNER_PROJECT_ID', 'your-gcp-project-id') instance_id = os.environ.get('SPANNER_INSTANCE_ID', 'your-spanner-instance-id') database_id = os.environ.get('SPANNER_DATABASE_ID', 'your-spanner-database-id') # For SQLAlchemy 1.4+ and 2.0, use 'spanner+spanner:///' # For SQLAlchemy 1.3, use 'spanner:///' db_url = f"spanner+spanner:///projects/{project_id}/instances/{instance_id}/databases/{database_id}" try: engine = create_engine(db_url) # Example: Define a table for demonstration metadata = MetaData() users = Table( 'users', metadata, Column('user_id', String(36), primary_key=True), Column('name', String(255), nullable=False), Column('age', Integer) ) # Create tables (DDL operations are not transactional in Spanner) with engine.connect() as connection: connection.execute(text("CREATE TABLE IF NOT EXISTS users (user_id STRING(36) NOT NULL, name STRING(255) NOT NULL, age INT64) PRIMARY KEY (user_id)")) connection.commit() print("Table 'users' ensured.") # Insert data with engine.begin() as connection: connection.execute(users.insert(), {"user_id": "user1", "name": "Alice", "age": 30}) connection.execute(users.insert(), {"user_id": "user2", "name": "Bob", "age": 24}) print("Data inserted.") # Query data with engine.connect() as connection: result = connection.execute(users.select().where(users.c.age > 25)).fetchall() print("Users older than 25:", result) except Exception as e: print(f"An error occurred: {e}") print("Ensure Cloud Spanner project, instance, and database are correctly configured and `GOOGLE_APPLICATION_CREDENTIALS` is set or default credentials are available.")