E-commerce MCP Server

"""Database models for e-commerce MCP server.""" from datetime import datetime from sqlalchemy import ( Column, Integer, String, Float, DateTime, ForeignKey, Text, Boolean, create_engine, ) from sqlalchemy.orm import declarative_base, relationship, sessionmaker Base = declarative_base() class Product(Base): """Product model for inventory management.""" __tablename__ = "products" id = Column(Integer, primary_key=True, autoincrement=True) name = Column(String(255), nullable=False) description = Column(Text, nullable=True) sku = Column(String(100), unique=True, nullable=False) price = Column(Float, nullable=False) cost = Column(Float, nullable=True) # For profit analysis quantity_in_stock = Column(Integer, default=0) category = Column(String(100), nullable=True) is_active = Column(Boolean, default=True) created_at = Column(DateTime, default=datetime.utcnow) updated_at = Column(DateTime, default=datetime.utcnow, onupdate=datetime.utcnow) order_items = relationship("OrderItem", back_populates="product") def to_dict(self): return { "id": self.id, "name": self.name, "description": self.description, "sku": self.sku, "price": self.price, "cost": self.cost, "quantity_in_stock": self.quantity_in_stock, "category": self.category, "is_active": self.is_active, "created_at": self.created_at.isoformat() if self.created_at else None, "updated_at": self.updated_at.isoformat() if self.updated_at else None, } class Customer(Base): """Customer model for customer management.""" __tablename__ = "customers" id = Column(Integer, primary_key=True, autoincrement=True) email = Column(String(255), unique=True, nullable=False) first_name = Column(String(100), nullable=False) last_name = Column(String(100), nullable=False) phone = Column(String(20), nullable=True) address = Column(Text, nullable=True) city = Column(String(100), nullable=True) state = Column(String(100), nullable=True) postal_code = Column(String(20), nullable=True) country = Column(String(100), nullable=True) is_active = Column(Boolean, default=True) created_at = Column(DateTime, default=datetime.utcnow) updated_at = Column(DateTime, default=datetime.utcnow, onupdate=datetime.utcnow) orders = relationship("Order", back_populates="customer") def to_dict(self): return { "id": self.id, "email": self.email, "first_name": self.first_name, "last_name": self.last_name, "phone": self.phone, "address": self.address, "city": self.city, "state": self.state, "postal_code": self.postal_code, "country": self.country, "is_active": self.is_active, "created_at": self.created_at.isoformat() if self.created_at else None, "updated_at": self.updated_at.isoformat() if self.updated_at else None, } class Order(Base): """Order model for order management.""" __tablename__ = "orders" id = Column(Integer, primary_key=True, autoincrement=True) customer_id = Column(Integer, ForeignKey("customers.id"), nullable=False) order_number = Column(String(50), unique=True, nullable=False) status = Column(String(50), default="pending") # pending, confirmed, shipped, delivered, cancelled total_amount = Column(Float, default=0.0) shipping_address = Column(Text, nullable=True) notes = Column(Text, nullable=True) created_at = Column(DateTime, default=datetime.utcnow) updated_at = Column(DateTime, default=datetime.utcnow, onupdate=datetime.utcnow) customer = relationship("Customer", back_populates="orders") items = relationship("OrderItem", back_populates="order", cascade="all, delete-orphan") def to_dict(self): return { "id": self.id, "customer_id": self.customer_id, "order_number": self.order_number, "status": self.status, "total_amount": self.total_amount, "shipping_address": self.shipping_address, "notes": self.notes, "created_at": self.created_at.isoformat() if self.created_at else None, "updated_at": self.updated_at.isoformat() if self.updated_at else None, "items": [item.to_dict() for item in self.items] if self.items else [], } class OrderItem(Base): """Order item model for line items in orders.""" __tablename__ = "order_items" id = Column(Integer, primary_key=True, autoincrement=True) order_id = Column(Integer, ForeignKey("orders.id"), nullable=False) product_id = Column(Integer, ForeignKey("products.id"), nullable=False) quantity = Column(Integer, nullable=False) unit_price = Column(Float, nullable=False) # Price at time of order order = relationship("Order", back_populates="items") product = relationship("Product", back_populates="order_items") def to_dict(self): return { "id": self.id, "order_id": self.order_id, "product_id": self.product_id, "product_name": self.product.name if self.product else None, "quantity": self.quantity, "unit_price": self.unit_price, "subtotal": self.quantity * self.unit_price, } def get_engine(db_path: str = "ecommerce.db"): """Create and return database engine.""" return create_engine(f"sqlite:///{db_path}", echo=False) def init_db(db_path: str = "ecommerce.db"): """Initialize the database with all tables.""" engine = get_engine(db_path) Base.metadata.create_all(engine) return engine def get_session(engine): """Create and return a new database session.""" Session = sessionmaker(bind=engine) return Session()