100 lines
No EOL
2.9 KiB
Python
100 lines
No EOL
2.9 KiB
Python
from alchemy_utils import CORE_ADDRESS_TABLE, CORE_USER_TABLE, DB_ENGINE
|
|
from declarative_orm import Address, User
|
|
from sqlalchemy import literal_column, select, text
|
|
from sqlalchemy.orm import Session
|
|
|
|
# stmt = select(CORE_USER_TABLE).where(CORE_USER_TABLE.c.name == "sandy")
|
|
|
|
# with DB_ENGINE.connect() as conn:
|
|
# for row in conn.execute(stmt):
|
|
# print(row)
|
|
|
|
# stmt = select(User).where(User.name=="sandy")
|
|
|
|
# with Session(DB_ENGINE) as session:
|
|
# for row in session.execute(statement=stmt):
|
|
# print(row)
|
|
|
|
# print(select(CORE_USER_TABLE.c.name, CORE_USER_TABLE.c.fullname))
|
|
# print(select(CORE_USER_TABLE.c["name", "fullname"]))
|
|
|
|
# print(select(User))
|
|
|
|
|
|
# with Session(DB_ENGINE) as session:
|
|
# row = session.execute(select(User)).first()
|
|
# print(row)
|
|
# rows = session.execute(select(User))
|
|
# print("-"*60)
|
|
# for row in rows:
|
|
# print(row)
|
|
|
|
# with Session(DB_ENGINE) as session:
|
|
# row = session.scalars(select(User)).first()
|
|
# print(row)
|
|
# rows = session.scalars(select(User))
|
|
# print("-"*60)
|
|
# for row in rows:
|
|
# print(row)
|
|
|
|
# with Session(DB_ENGINE) as session:
|
|
# row = session.execute(select(User.name, User.fullname))
|
|
# for r in row:
|
|
# print(r)
|
|
|
|
|
|
# with Session(DB_ENGINE) as session:
|
|
# result = session.execute(
|
|
# select(User.name, Address).where(User.id == Address.user_id).order_by(Address.id)
|
|
# ).all()
|
|
# print(result)
|
|
|
|
# stmt = select(
|
|
# ("Username: " + CORE_USER_TABLE.c.name).label("username")
|
|
# ).order_by(CORE_USER_TABLE.c.name)
|
|
# with DB_ENGINE.connect() as conn:
|
|
# for row in conn.execute(stmt):
|
|
# print(f"{row.username}")
|
|
|
|
# stmt = select(text("'some phrase'"), CORE_USER_TABLE.c.name).order_by(CORE_USER_TABLE.c.name)
|
|
# with DB_ENGINE.connect() as conn:
|
|
# print(conn.execute(stmt).all())
|
|
|
|
# stmt = select(literal_column("'some phrase'").label("p"), CORE_USER_TABLE.c.name).order_by(
|
|
# CORE_USER_TABLE.c.name
|
|
# )
|
|
# with DB_ENGINE.connect() as conn:
|
|
# for row in conn.execute(stmt):
|
|
# print(f"{row.p}, {row.name}")
|
|
|
|
# print(select(CORE_USER_TABLE).where(CORE_USER_TABLE.c.name=="patrick"))
|
|
|
|
# print(
|
|
# select(CORE_ADDRESS_TABLE.c.email_address)
|
|
# .where(CORE_USER_TABLE.c.name == "patrick")
|
|
# .where(CORE_ADDRESS_TABLE.c.user_id == CORE_USER_TABLE.c.id)
|
|
# )
|
|
|
|
# from sqlalchemy import and_, or_
|
|
# print(
|
|
# select(Address.email_address).where(
|
|
# and_(
|
|
# or_(User.name == "patrick", User.name == "sandy"),
|
|
# Address.user_id == User.id,
|
|
# )
|
|
# )
|
|
# )
|
|
|
|
# print(select(CORE_USER_TABLE.c.name))
|
|
|
|
# print(
|
|
# select(CORE_USER_TABLE.c.name, CORE_ADDRESS_TABLE.c.email_address).join_from(
|
|
# CORE_USER_TABLE, CORE_ADDRESS_TABLE
|
|
# )
|
|
# )
|
|
|
|
print(
|
|
select(CORE_ADDRESS_TABLE.c.email_address)
|
|
.select_from(CORE_USER_TABLE)
|
|
.join(CORE_ADDRESS_TABLE, CORE_USER_TABLE.c.id == CORE_ADDRESS_TABLE.c.user_id)
|
|
) |