Soy nuevo en FastAPI y estoy trabajando en la inserción de datos en la base de datos MY SQL utilizando el método Fast API POST. Creé un conjunto de códigos de muestra para crear un esquema e inserté los datos individuales en la tabla MY SQL usando el siguiente código de muestra.
Enlace de referencia: https://codingnomads.co/blog/python-fastapi-tutorial
from fastapi import FastAPI, Depends from pydantic import BaseModel from typing import Optional, List from sqlalchemy import create_engine from sqlalchemy.orm import declarative_base, sessionmaker, Session from sqlalchemy import Boolean, Column, Float, String, Integer app = FastAPI() # SqlAlchemy Setup SQLALCHEMY_DATABASE_URL = 'sqlite+pysqlite:///./db.sqlite3:' engine = create_engine(SQLALCHEMY_DATABASE_URL, echo=True, future=True) SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine) Base = declarative_base() def get_db(): db = SessionLocal() try: yield db finally: db.close() # A SQLAlchemny ORM Place class DBPlace(Base): __tablename__ = 'places' id = Column(Integer, primary_key=True, index=True) name = Column(String(50)) description = Column(String, nullable=True) coffee = Column(Boolean) wifi = Column(Boolean) food = Column(Boolean) lat = Column(Float) lng = Column(Float) Base.metadata.create_all(bind=engine) # A Pydantic Place class Place(BaseModel): name: str description: Optional[str] = None coffee: bool wifi: bool food: bool lat: float lng: float class Config: orm_mode = True # Methods for interacting with the database def get_place(db: Session, place_id: int): return db.query(DBPlace).where(DBPlace.id == place_id).first() def get_places(db: Session): return db.query(DBPlace).all() def create_place(db: Session, place: Place): db_place = DBPlace(**place.dict()) db.add(db_place) db.commit() db.refresh(db_place) return db_place # Routes for interacting with the API @app.post('/places/', response_model=Place) def create_places_view(place: Place, db: Session = Depends(get_db)): db_place = create_place(db, place) return db_place @app.get('/places/', response_model=List[Place]) def get_places_view(db: Session = Depends(get_db)): return get_places(db) @app.get('/place/{place_id}') def get_place_view(place_id: int, db: Session = Depends(get_db)): return get_place(db, place_id) @app.get('/') async def root(): return {'message': 'Hello World!'}Pero necesito analizar una lista de matrices en la solicitud de create_places_view e insertar varios valores a la vez. Entonces, ¿cómo lograr esto en Fast API? ¿Algún consejo/ayuda? Gracias
Ejemplo:
[{name: str description: Optional[str] = None coffee: bool wifi: bool food: bool lat: float lng: float }, {name: str description: Optional[str] = None coffee: bool wifi: bool food: bool lat: float lng: float }]
He enlazado el valor para crear un usuario. Al enviar la solicitud de publicación en la lista de dict.
# Fast API to create/insert a new user @app.post('/create_place_view/', responses={404: {"model": Message}}) async def create_place(user: List[Place], db: Session = Depends(get_db)): if not user: return JSONResponse(status_code=404, content={"message": "Place details not found"}) else: places_list = [] for places in place: db_place = create_places(db, places) place_json = places.dict() places_list.append(places_json) return JSONResponse(content=places_list)