#!/usr/bin/env python from sqlalchemy import create_engine from sqlalchemy import text from sqlalchemy import Table, Column, Integer, String from sqlalchemy import ForeignKey from sqlalchemy import select from sqlalchemy import insert from sqlalchemy.orm import Session from sqlalchemy.orm import declarative_base from sqlalchemy.orm import relationship Base = declarative_base() engine = create_engine("sqlite+pysqlite:///:memory:", echo=True, future=True) class User(Base): __tablename__ = 'user_account' id = Column(Integer, primary_key=True) name = Column(String(30)) fullname = Column(String) addresses = relationship("Address", back_populates="user") def __repr__(self): return ( f"User(id={self.id!r}, name={self.name!r}, " f"fullname={self.fullname!r})" ) class Address(Base): __tablename__ = 'address' id = Column(Integer, primary_key=True) email_address = Column(String, nullable=False) user_id = Column(Integer, ForeignKey('user_account.id')) user = relationship("User", back_populates="addresses") def __repr__(self): return f"Address(id={self.id!r}, email_address={self.email_address!r})" Base.metadata.create_all(engine) squidward = User(name="squidward", fullname="Squidward Tentacles") krabs = User(name="ehkrabs", fullname="Eugene H. Krabs") session = Session(engine) session.add(squidward) session.add(krabs) session.commit() u1 = User(name='pkrabs', fullname='Pearl Krabs') a1 = Address(email_address="pearl.krabs@gmail.com") u1.addresses.append(a1) a2 = Address(email_address="pearl@aol.com", user=u1) session.add(u1) session.add(a1) session.add(a2) session.commit() # with engine.connect() as conn: # conn.execute(text("CREATE TABLE some_table (x int, y int)")) # conn.execute( # text("INSERT INTO some_table (x, y) VALUES (:x, :y)"), # [{"x": 1, "y": 1}, {"x": 2, "y": 4}] # ) # conn.commit() # # with engine.begin() as conn: # conn.execute( # text("INSERT INTO some_table (x, y) VALUES (:x, :y)"), # [{"x": 6, "y": 8}, {"x": 9, "y": 10}] # ) # # # with engine.connect() as conn: # # result = conn.execute(text("SELECT x, y FROM some_table")) # # for row in result: # # print(f"x: {row.x} y: {row.y}") # # # stmt = text( # "SELECT x, y FROM some_table WHERE y > :y ORDER BY x, y").bindparams(y=6) # # with Session(engine) as session: # result = session.execute(stmt) # for row in result: # print(f"x: {row.x} y: {row.y}")