marshmallow-sqlalchemy

import datetime from sqlalchemy import create_engine, Column, Integer, String, DateTime from sqlalchemy.orm import sessionmaker, declarative_base from marshmallow_sqlalchemy import SQLAlchemyAutoSchema from marshmallow import fields # 1. Define SQLAlchemy Base and Model Base = declarative_base() class User(Base): __tablename__ = 'users' id = Column(Integer, primary_key=True) name = Column(String(50), nullable=False) email = Column(String(100), unique=True, nullable=False) created_at = Column(DateTime, default=datetime.datetime.now) def __repr__(self): return f"<User(id={self.id}, name='{self.name}')>" # 2. Set up database (in-memory SQLite for example) engine = create_engine('sqlite:///:memory:') Base.metadata.create_all(engine) Session = sessionmaker(bind=engine) session = Session() # 3. Define Marshmallow Schema for the User model class UserSchema(SQLAlchemyAutoSchema): class Meta: model = User load_instance = True # Optional: Allows loading data into existing instances sqla_session = session # Essential for relationship handling # fields = ('id', 'name', 'email', 'created_at') # Explicit fields # Example of overriding a field (optional) created_at = fields.DateTime(format='%Y-%m-%d %H:%M:%S') # 4. Usage Example user_data = { 'name': 'Alice Smith', 'email': 'alice@example.com' } # Create a schema instance user_schema = UserSchema() # Deserialize (load) data to create a new User object new_user = user_schema.load(user_data, session=session) session.add(new_user) session.commit() print(f"Created user: {new_user}") # Serialize (dump) an existing User object serialized_user = user_schema.dump(new_user) print(f"Serialized user: {serialized_user}") # Update an existing user update_data = {'name': 'Alicia Smith'} alice = session.query(User).filter_by(email='alice@example.com').first() updated_user = user_schema.load(update_data, instance=alice, session=session) session.add(updated_user) session.commit() print(f"Updated user: {updated_user}") session.close()