83 lines
2.3 KiB
Python
83 lines
2.3 KiB
Python
from typing import List, Optional
|
|
from dotenv import load_dotenv
|
|
|
|
from sqlalchemy import String, create_engine, Boolean, text
|
|
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, relationship
|
|
import os
|
|
|
|
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')
|
|
|
|
# engine.connect()
|
|
|
|
|
|
# class Base(DeclarativeBase):
|
|
# pass
|
|
|
|
|
|
# class User(Base):
|
|
# __tablename__ = "user_account"
|
|
# id: Mapped[int] = mapped_column(primary_key=True)
|
|
# username: Mapped[str] = mapped_column(String(30))
|
|
# first_name: Mapped[str] = mapped_column(String(50))
|
|
# last_name: Mapped[Optional[str]] = mapped_column(String(50))
|
|
# is_coach: Mapped[bool] = mapped_column(Boolean())
|
|
|
|
# def __repr__(self) -> str:
|
|
# return f"User(id={self.id!r}, name={self.name!r}, first_name={self.first_name!r}, is_coach={self.is_coach})"
|
|
|
|
# Base.metadata.create_all(engine)
|
|
|
|
# with engine.connect() as conn:
|
|
# result = conn.execute(text("select 'hello world'"))
|
|
# print(result.all())
|
|
|
|
# Commiting style one
|
|
|
|
# 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()
|
|
|
|
# Commiting style two
|
|
|
|
# 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}]
|
|
# )
|
|
|
|
# Managing Result object
|
|
# 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}")
|
|
|
|
# Sending parameters
|
|
# with engine.connect() as conn:
|
|
# result = conn.execute(text("SELECT x, y FROM some_table where y > :y"), {"y": 20})
|
|
# for row in result:
|
|
# print(f"x: {row.x} y: {row.y}")
|
|
|
|
# Sending multiple params
|
|
|
|
# with engine.connect() as conn:
|
|
# conn.execute(
|
|
# text("INSERT INTO some_table (x, y) VALUES (:x, :y)"),
|
|
# [{"x": 11, "y": 12}, {"x": 13, "y": 14}],
|
|
# )
|
|
# conn.commit()
|
|
|
|
with engine.connect() as conn:
|
|
conn.execute(
|
|
text("DROP TABLE some_table")
|
|
)
|
|
conn.commit()
|