48 lines
1.6 KiB
Python
48 lines
1.6 KiB
Python
from sqlalchemy import create_engine, Column, Integer, String, Float, DateTime
|
|
from sqlalchemy.ext.declarative import declarative_base
|
|
from sqlalchemy.orm import sessionmaker
|
|
from sqlalchemy.sql import func
|
|
import os
|
|
|
|
# Database URL from environment variable
|
|
DATABASE_URL = os.getenv("DATABASE_URL")
|
|
if not DATABASE_URL:
|
|
# Fallback to constructing from individual variables
|
|
POSTGRES_USER = os.getenv("POSTGRES_USER", "fastapi_user")
|
|
POSTGRES_PASSWORD = os.getenv("POSTGRES_PASSWORD", "fastapi_password")
|
|
POSTGRES_HOST = os.getenv("POSTGRES_HOST", "postgres")
|
|
POSTGRES_PORT = os.getenv("POSTGRES_PORT", "5432")
|
|
POSTGRES_DB = os.getenv("POSTGRES_DB", "fastapi_db")
|
|
DATABASE_URL = f"postgresql://{POSTGRES_USER}:{POSTGRES_PASSWORD}@{POSTGRES_HOST}:{POSTGRES_PORT}/{POSTGRES_DB}"
|
|
|
|
# Create SQLAlchemy engine
|
|
engine = create_engine(DATABASE_URL)
|
|
|
|
# Create SessionLocal class
|
|
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
|
|
|
|
# Create Base class
|
|
Base = declarative_base()
|
|
|
|
# Database model
|
|
class Item(Base):
|
|
__tablename__ = "items"
|
|
|
|
id = Column(Integer, primary_key=True, index=True)
|
|
name = Column(String, nullable=False)
|
|
description = Column(String, nullable=True)
|
|
price = Column(Float, nullable=False)
|
|
created_at = Column(DateTime(timezone=True), server_default=func.now())
|
|
updated_at = Column(DateTime(timezone=True), onupdate=func.now())
|
|
|
|
# Dependency to get database session
|
|
def get_db():
|
|
db = SessionLocal()
|
|
try:
|
|
yield db
|
|
finally:
|
|
db.close()
|
|
|
|
# Create tables
|
|
def create_tables():
|
|
Base.metadata.create_all(bind=engine) |