70 lines
2.1 KiB
Python
70 lines
2.1 KiB
Python
import os
|
|
from typing import List, Optional
|
|
|
|
from dotenv import load_dotenv
|
|
from sqlalchemy import (Boolean, Column, ForeignKey, Integer, MetaData, String,
|
|
Table, create_engine, text, insert, select)
|
|
from sqlalchemy.orm import (DeclarativeBase, Mapped, Session, mapped_column,
|
|
relationship)
|
|
|
|
load_dotenv()
|
|
|
|
DB_PASS = os.getenv("DOCKER_POSTGRES_PASS")
|
|
|
|
# Create engine for connecting to postgres test db
|
|
engine = create_engine(
|
|
f'postgresql+psycopg2://postgres:{DB_PASS}@localhost:5432/postgres')
|
|
|
|
metadata_object = MetaData()
|
|
|
|
# Declaring user table
|
|
user_table = Table(
|
|
"user_account",
|
|
metadata_object,
|
|
Column("id", Integer, primary_key=True),
|
|
Column("name", String(30)),
|
|
Column("fullname", String),
|
|
)
|
|
|
|
# Declaring address table
|
|
address_table = Table(
|
|
"address",
|
|
metadata_object,
|
|
Column("id", Integer, primary_key=True),
|
|
Column("user_id", ForeignKey("user_account.id"), nullable=False),
|
|
Column("email_address", String, nullable=False)
|
|
)
|
|
|
|
# metadata_object.create_all(engine)
|
|
|
|
# stmt = insert(user_table).values(name="spongebob", fullname="Spongebob squarepants").returning(user_table.c.id, user_table.c.name)
|
|
# select_statement = select(user_table.c.id, user_table.c.name + "@google.com")
|
|
# insert_statement = insert(address_table).from_select(
|
|
# ["user_id", "email_address"], select_statement
|
|
# ).returning(address_table.c.id, address_table.c.email_address)
|
|
|
|
# with engine.connect() as conn:
|
|
# result = conn.execute(insert_statement)
|
|
# conn.commit()
|
|
# for row in result:
|
|
# print(row)
|
|
|
|
# with engine.connect() as conn:
|
|
# result = conn.execute(
|
|
# insert(user_table),
|
|
# [
|
|
# {"name": "sandy", "fullname": "Sandy Cheeks"},
|
|
# {"name": "patrick", "fullname": "Patrick Star"},
|
|
# ],
|
|
# )
|
|
# conn.commit()
|
|
|
|
# insert_stmt = insert(address_table).returning(
|
|
# address_table.c.id, address_table.c.email_address
|
|
# )
|
|
# print(insert_stmt)
|
|
|
|
# with engine.connect() as conn:
|
|
# result = conn.execute(stmt)
|
|
# conn.commit()
|
|
# print(result)
|