sqlalchemy_spanner_crud_quickstart_with_orm.py

1from sqlalchemy import create_engine, Column, Integer, String, MetaData, Table
2from sqlalchemy.orm import declarative_base, sessionmaker
3
4# Replace these values with your actual Spanner instance details
5project_id = "your-project-id"
6instance_id = "your-instance-id"
7database_id = "your-database-id"
8
9# Connection string format: spanner:///projects/<project>/instances/<instance>/databases/<database>
10connection_url = f"spanner:///projects/{project_id}/instances/{instance_id}/databases/{database_id}"
11
12# Create the engine
13engine = create_engine(connection_url)
14Base = declarative_base()
15
16# Define a sample Model
17class User(Base):
18    __tablename__ = "users"
19    user_id = Column(Integer, primary_key=True)
20    user_name = Column(String(16), nullable=False)
21
22# Create the table in Spanner
23Base.metadata.create_all(engine)
24
25# Create a session and perform operations
26Session = sessionmaker(bind=engine)
27session = Session()
28
29# Insert data
30new_user = User(user_id=1, user_name="SpannerUser")
31session.add(new_user)
32session.commit()
33
34# Query data
35user = session.query(User).filter_by(user_name="SpannerUser").first()
36print(f"Retrieved User: {user.user_id}, {user.user_name}")
37
38# Close the session
39session.close()