sqlalchemy.py 文件源码

def get_or_create(db, model, create_method: str='',
                  create_method_kwargs=None, **kwargs):
    try:
        return db.query(model).filter_by(**kwargs).one()
    except NoResultFound:
        pass

    kwargs.update(create_method_kwargs or {})
    created = getattr(model, create_method, model)(**kwargs)
    try:
        db.add(created)
        db.flush()
        return created
    except IntegrityError:
        pass

    db.rollback()
    return db.query(model).filter_by(**kwargs).one()