import json from sqlalchemy import create_engine, Column, Integer, String, Text, DateTime, ForeignKey, func from sqlalchemy.ext.declarative import declarative_base from sqlalchemy.orm import sessionmaker, relationship from sqlalchemy.types import TypeDecorator, TEXT # Custom type for JSON storage class JSONText(TypeDecorator): impl = TEXT def process_bind_param(self, value, dialect): if value is not None: return json.dumps(value) return None def process_result_value(self, value, dialect): if value is not None: return json.loads(value) return None # Database connection URL (using SQLite) DATABASE_URL = "sqlite:///./api_proxy.db" # Create SQLAlchemy engine engine = create_engine(DATABASE_URL, connect_args={"check_same_thread": False}) # Create a configured "Session" class SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine) # Base class for declarative models Base = declarative_base() # Define the KeyCategory model class KeyCategory(Base): __tablename__ = "key_categories" id = Column(Integer, primary_key=True, index=True) name = Column(String, unique=True, index=True, nullable=False) type = Column(String, nullable=False) tags = Column(JSONText) # Stored as JSON string metadata_ = Column(JSONText, name="metadata") # Map to 'metadata' column in DB api_keys = relationship("APIKey", back_populates="category") created_at = Column(DateTime, server_default=func.now()) updated_at = Column(DateTime, server_default=func.now(), onupdate=func.now()) # Define the APIKey model class APIKey(Base): __tablename__ = "api_keys" id = Column(Integer, primary_key=True, index=True) value = Column(String, nullable=False) category_id = Column(Integer, ForeignKey("key_categories.id"), nullable=False) status = Column(String, default="active", nullable=False) # 'active', 'inactive' usage_count = Column(Integer, default=0, nullable=False) last_used = Column(DateTime) metadata_ = Column(JSONText, name="metadata") # Map to 'metadata' column in DB category = relationship("KeyCategory", back_populates="api_keys") created_at = Column(DateTime, server_default=func.now()) updated_at = Column(DateTime, server_default=func.now(), onupdate=func.now()) # Function to create database tables def create_db_tables(): Base.metadata.create_all(bind=engine) # Dependency to get DB session def get_db(): db = SessionLocal() try: yield db finally: db.close()