90 lines
2.9 KiB
Python
90 lines
2.9 KiB
Python
Sure, here's an example of a FastAPI endpoint that saves data to a database using SQLAlchemy and Pydantic:
|
|
|
|
```python
|
|
from typing import List
|
|
from fastapi import FastAPI, HTTPException
|
|
from pydantic import BaseModel
|
|
from sqlalchemy import Column, Integer, String, create_engine
|
|
from sqlalchemy.orm import sessionmaker
|
|
from sqlalchemy.ext.declarative import declarative_base
|
|
|
|
# Create the database engine and session
|
|
engine = create_engine('sqlite:///books.db', echo=True)
|
|
Session = sessionmaker(bind=engine)
|
|
session = Session()
|
|
|
|
# Create the base class for SQLAlchemy models
|
|
Base = declarative_base()
|
|
|
|
# Define the SQLAlchemy model
|
|
class Book(Base):
|
|
__tablename__ = 'books'
|
|
|
|
id = Column(Integer, primary_key=True)
|
|
title = Column(String)
|
|
author = Column(String)
|
|
|
|
def __repr__(self):
|
|
return f"Book(id={self.id}, title='{self.title}', author='{self.author}')"
|
|
|
|
# Create the database table
|
|
Base.metadata.create_all(engine)
|
|
|
|
# Define the Pydantic model
|
|
class BookSchema(BaseModel):
|
|
title: str
|
|
author: str
|
|
|
|
class Config:
|
|
orm_mode = True
|
|
|
|
# Create the FastAPI app
|
|
app = FastAPI()
|
|
|
|
# Define the POST endpoint
|
|
@app.post('/books', response_model=List[BookSchema])
|
|
def create_book(book: BookSchema):
|
|
# Create a new book instance
|
|
new_book = Book(title=book.title, author=book.author)
|
|
|
|
# Add the new book to the session
|
|
session.add(new_book)
|
|
|
|
# Commit the changes to the database
|
|
session.commit()
|
|
|
|
# Retrieve all books from the database
|
|
books = session.query(Book).all()
|
|
|
|
# Convert the SQLAlchemy models to Pydantic models
|
|
return [BookSchema.from_orm(book) for book in books]
|
|
|
|
# Run the app
|
|
if __name__ == '__main__':
|
|
import uvicorn
|
|
uvicorn.run(app, host='0.0.0.0', port=8000)
|
|
```
|
|
|
|
Here's a breakdown of the code:
|
|
|
|
1. We define the SQLAlchemy model `Book` with `id`, `title`, and `author` columns.
|
|
2. We create the database table using `Base.metadata.create_all(engine)`.
|
|
3. We define the Pydantic model `BookSchema` with `title` and `author` fields.
|
|
4. We create the FastAPI app instance.
|
|
5. We define the `POST` endpoint `/books` that accepts a `BookSchema` instance as input.
|
|
6. Inside the endpoint function, we create a new `Book` instance using the data from the `BookSchema` instance.
|
|
7. We add the new `Book` instance to the session and commit the changes to the database.
|
|
8. We retrieve all books from the database using `session.query(Book).all()`.
|
|
9. We convert the SQLAlchemy models to Pydantic models using `BookSchema.from_orm(book)`.
|
|
10. We return the list of `BookSchema` instances as the response.
|
|
|
|
To run the app, execute `uvicorn main:app --reload` in your terminal, and then you can send a POST request to `http://localhost:8000/books` with a JSON payload like:
|
|
|
|
```json
|
|
{
|
|
"title": "The Great Gatsby",
|
|
"author": "F. Scott Fitzgerald"
|
|
}
|
|
```
|
|
|
|
This will create a new book in the database and return a list of all books in the database. |