63 lines
2.4 KiB
Python
63 lines
2.4 KiB
Python
Sure, here's an example of a FastAPI endpoint that saves people of France to the database using SQLAlchemy and Pydantic:
|
|
|
|
```python
|
|
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.people_model import *
|
|
from app.api.schemas.people_schema import *
|
|
app = FastAPI()
|
|
|
|
# SQLAlchemy setup
|
|
Base = declarative_base()
|
|
engine = create_engine('sqlite:///people.db', connect_args={'check_same_thread': False})
|
|
Session = sessionmaker(bind=engine)
|
|
|
|
# SQLAlchemy model
|
|
|
|
class Config:
|
|
orm_mode = True
|
|
|
|
# FastAPI endpoint
|
|
@app.post('/people', response_model=PersonSchema)
|
|
def create_person(person: PersonSchema):
|
|
session = Session()
|
|
try:
|
|
new_person = Person(name=person.name, city=person.city)
|
|
session.add(new_person)
|
|
session.commit()
|
|
return new_person
|
|
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 SQLite database and define a `Person` model with `name` and `city` columns.
|
|
3. We create a Pydantic schema `PersonSchema` to validate the request data.
|
|
4. We define a FastAPI endpoint `/people` with the `POST` method and use the `PersonSchema` as the request body and response model.
|
|
5. Inside the endpoint function `create_person`, we create a new SQLAlchemy session and try to add a new `Person` instance to the database using the provided `name` and `city` from the request data.
|
|
6. If the operation is successful, we commit the changes and return the newly created `Person` instance.
|
|
7. If an exception occurs, we roll back the transaction and raise an `HTTPException` 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 `http://localhost:8000/people` with a JSON payload containing the `name` and `city` fields, like this:
|
|
|
|
```json
|
|
{
|
|
"name": "John Doe",
|
|
"city": "Paris"
|
|
}
|
|
```
|
|
|
|
The endpoint will create a new person in the database and return the created person object as a JSON response.
|
|
|
|
Note: Make sure you have the required dependencies installed (`fastapi`, `pydantic`, and `sqlalchemy`). |