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()