sqlalchemy_spanner_crud_operations_quickstart.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 Cloud Spanner connection details
5# Format: spanner:///projects/<project>/instances/<instance>/databases/<database>
6engine = create_engine("spanner:///projects/your-project-id/instances/your-instance-id/databases/your-database-id")
7
8Base = declarative_base()
9
10class User(Base):
11    __tablename__ = "users"
12    user_id = Column(Integer, primary_key=True)
13    user_name = Column(String(16), nullable=False)
14
15# Create the table in Spanner
16Base.metadata.create_all(engine)
17
18# Create a session and interact with the database
19Session = sessionmaker(bind=engine)
20session = Session()
21
22# Insert a new record
23new_user = User(user_id=1, user_name="SpannerUser")
24session.add(new_user)
25session.commit()
26
27# Query the record
28user = session.query(User).filter_by(user_name="SpannerUser").first()
29print(f"Found user: {user.user_id}, {user.user_name}")
30
31# Close the session
32session.close()