Quiero exponer un método API que recibe datos en una solicitud POST (para una API de registro beta) y actualizar con nuevos valores cuando ya hay un modelo coincidente. ¿Cuál es la mejor manera de lograr esto? Actualmente estoy haciendo esto (algo simplificado):
Mi modelo ORM (SqlAlchemy):
class BetaORM(Base): __tablename__ = "betasignup" email = Column(EmailType, primary_key=True) fullname = Column(String, unique=False, index=False, nullable=True)Mi modelo Pydantic:
class BetaCreate(BaseModel): email: EmailStr fullname: Optional[str]Mi método de publicación FastAPI:
@app.post("/beta_signup") def post_beta_signup(beta: schemas.BetaCreate, db: Session = Depends(get_db)): return create_beta_signup(db=db,signup=beta)Y el método CRUD que he escrito:
def create_beta_signup(db: Session, signup: schemas.BetaCreate): db_beta = schemas.BetaORM(**signup.dict()) ret_obj = db.merge(db_beta) db.add(ret_obj) db.commit() return ret_objUn problema con el uso de merge () es que se basa en la coincidencia con la clave principal, directamente con la dirección de correo electrónico. una restricción única en la dirección de correo electrónico a nivel de base de datos.
Basado en la aplicación de demostración FastAPI sqlalchemy , así es como resolvió este problema:
def update_user(db: Session, user: PydanticUserUpdate): """ Using a new update method seen in FastAPI https://github.com/tiangolo/fastapi/pull/2665 Simple, does not need each attribute to be updated individually Uses python in built functionality... preferred to the pydintic related method """ # get the existing data db_user = db.query(User).filter(User.id == user.id).one_or_none() if db_user is None: return None # Update model class variable from requested fields for var, value in vars(user).items(): setattr(db_user, var, value) if value else None db_user.modified = modified_now db.add(db_user) db.commit() db.refresh(db_user) return db_userEste enfoque utiliza la definición del modelo declarativo de SQLAlchemy (no la definición imperativa que utiliza Gabriel Cappelli)
Usé este enfoque para hacer una actualización en crud en una aplicación basada en FastAPI. Los datos existentes no se modifican y los nuevos valores de actualización se sobrescriben. La fecha y hora modificada se actualiza (pero para facilitar la prueba, este valor es fijo).
Espero eso ayude. (Pasé demasiado tiempo resolviendo esto).
Si está utilizando MySQL y SQLAlchemy> = 1.2, puede usar INSERT...ON DUPLICATE KEY UPDATE usando SQLAlchemy.
from sqlalchemy.dialects.mysql import insert insert_stmt = insert(my_table).values( id='some_existing_id', data='inserted value') on_duplicate_key_stmt = insert_stmt.on_duplicate_key_update( data=insert_stmt.inserted.data, status='U' ) conn.execute(on_duplicate_key_stmt)Más información en los documentos