Spaces:
Sleeping
Sleeping
File size: 2,567 Bytes
b7791c2 7183ec8 b7791c2 7183ec8 b7791c2 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 |
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()
|