69 lines
2.3 KiB
Python
69 lines
2.3 KiB
Python
Sure, here's an example of a FastAPI endpoint that adds cities to the database using SQLAlchemy models and Pydantic schemas:
|
|
|
|
```python
|
|
from typing import List
|
|
from fastapi import FastAPI, HTTPException
|
|
from pydantic import BaseModel
|
|
from sqlalchemy import Column, Integer, String
|
|
from sqlalchemy.ext.declarative import declarative_base
|
|
from sqlalchemy import create_engine
|
|
from sqlalchemy.orm import sessionmaker
|
|
|
|
from app.api.models.cities_model import *
|
|
from app.api.schemas.cities_schema import *
|
|
app = FastAPI()
|
|
|
|
# SQLAlchemy setup
|
|
Base = declarative_base()
|
|
engine = create_engine('sqlite:///cities.db', echo=True)
|
|
Session = sessionmaker(bind=engine)
|
|
|
|
# SQLAlchemy model
|
|
|
|
class Config:
|
|
orm_mode = True
|
|
|
|
# POST endpoint to add cities
|
|
@app.post("/cities", response_model=List[CitySchema])
|
|
def add_cities(cities: List[CitySchema]):
|
|
session = Session()
|
|
|
|
try:
|
|
db_cities = [City(name=city.name, country=city.country) for city in cities]
|
|
session.add_all(db_cities)
|
|
session.commit()
|
|
return db_cities
|
|
except Exception as e:
|
|
session.rollback()
|
|
raise HTTPException(status_code=500, detail=str(e))
|
|
finally:
|
|
session.close()
|
|
```
|
|
|
|
Here's a breakdown of the code:
|
|
|
|
1. We import the necessary modules and create a FastAPI instance.
|
|
2. We set up SQLAlchemy with a `City` model that has `id`, `name`, and `country` columns.
|
|
3. We create a Pydantic `CitySchema` model that will be used for request and response validation.
|
|
4. We define a `POST` endpoint at `/cities` that accepts a list of `CitySchema` objects in the request body.
|
|
5. Inside the endpoint function, we create a SQLAlchemy session and try to add the cities to the database.
|
|
6. If the operation is successful, we commit the changes and return the list of added cities.
|
|
7. If an exception occurs, we roll back the transaction and raise an HTTP exception with a 500 status code and the error message.
|
|
8. Finally, we close the SQLAlchemy session.
|
|
|
|
To use this endpoint, you can send a `POST` request to `/cities` with a JSON payload containing a list of city objects, like this:
|
|
|
|
```json
|
|
[
|
|
{
|
|
"name": "New York",
|
|
"country": "USA"
|
|
},
|
|
{
|
|
"name": "London",
|
|
"country": "UK"
|
|
}
|
|
]
|
|
```
|
|
|
|
The endpoint will add these cities to the database and return the list of added cities as a response. |