sqlalchemy-mixins

from sqlalchemy import create_engine, Column, Integer, String from sqlalchemy.orm import sessionmaker, declarative_base from sqlalchemy_mixins import ActiveRecordMixin import os # Setup SQLAlchemy engine and session (using in-memory sqlite for example) # For a real app, you might get this from an environment variable database_url = os.environ.get('DATABASE_URL', 'sqlite:///test.db') engine = create_engine(database_url) Session = sessionmaker(bind=engine) session = Session() # Define the declarative base for models Base = declarative_base() # Define a User model inheriting from Base and ActiveRecordMixin class User(Base, ActiveRecordMixin): __tablename__ = 'users' id = Column(Integer, primary_key=True) name = Column(String) email = Column(String, unique=True) def __repr__(self): return f"<User(id={self.id}, name='{self.name}')>" # Create database tables Base.metadata.create_all(engine) # Set the session for ActiveRecordMixin for this model User.set_session(session) # --- Usage Examples --- # 1. Create a new user (commit=True is available from v2.0.5) user1 = User.create(name='Alice', email='alice@example.com', commit=True) print(f"Created user: {user1}") # 2. Find a user by ID found_user = User.find(user1.id) print(f"Found user by ID: {found_user}") # 3. Find a user by attributes (Django-like filter) filtered_user = User.where(name='Alice').first() print(f"Found user by name: {filtered_user}") # 4. Update a user user1.update(name='Alicia', commit=True) print(f"Updated user: {user1}") # 5. Get all users all_users = User.all() print(f"Total users: {len(all_users)}") # 6. Delete a user user1.delete(commit=True) print(f"User {user1.name} deleted.") # Clean up and close the session session.close()