FastAPI in 5 minutes

Here’s a step-by-step guide for getting started with FastAPI with a code example and sample database schema.

Step 1: Install FastAPI and its dependencies

First, you’ll need to install FastAPI and its dependencies. You can do this using pip:

pip install fastapi uvicorn

Step 2: Create a database schema

Next, let’s create a simple database schema that we can use for our example application. Here’s an example schema for a blog:

CREATE TABLE posts (
  id SERIAL PRIMARY KEY,
  title VARCHAR(255) NOT NULL,
  content TEXT NOT NULL,
  created_at TIMESTAMP DEFAULT NOW()
);

CREATE TABLE comments (
  id SERIAL PRIMARY KEY,
  post_id INTEGER NOT NULL,
  author VARCHAR(255) NOT NULL,
  content TEXT NOT NULL,
  created_at TIMESTAMP DEFAULT NOW()
);

Step 3: Create a database connection

We’ll use the psycopg2 library to connect to our Postgres database. Here’s an example of how to create a connection:

import psycopg2

conn = psycopg2.connect(
    dbname="your_database_name",
    user="your_username",
    password="your_password",
    host="your_host",
    port="your_port"
)

Step 4: Create a FastAPI application

Now that we have our database schema and connection, we can create a FastAPI application. Here’s an example of how to create a basic FastAPI application:

from fastapi import FastAPI

app = FastAPI()

@app.get("/")
def read_root():
    return {"Hello": "World"}

Step 5: Create API endpoints

Next, let’s create some API endpoints that allow us to interact with our database. Here’s an example of how to create an endpoint to retrieve all blog posts:

from fastapi import FastAPI
from typing import List

app = FastAPI()

@app.get("/")
def read_root():
    return {"Hello": "World"}

@app.get("/posts")
def get_posts():
    cursor = conn.cursor()
    cursor.execute("SELECT id, title, content, created_at FROM posts")
    results = cursor.fetchall()
    cursor.close()
    return [{"id": result[0], "title": result[1], "content": result[2], "created_at": result[3]} for result in results]

Step 6: Run the application

Finally, let’s run our application using the uvicorn server:

uvicorn main:app --reload

FastAPI Technical Details

Here are some technical details about FastAPI that you may find useful:

Fast API with ORM

FastAPI can work seamlessly with Object-Relational Mapping (ORM) tools in Python like SQLAlchemy, Tortoise ORM, and others. ORM tools provide an abstraction layer on top of the database, allowing developers to interact with the database using Python objects instead of writing SQL queries.

Here is an example of using FastAPI with SQLAlchemy ORM:

First, let’s install the required dependencies:

!pip install fastapi
!pip install uvicorn[standard]
!pip install sqlalchemy

Next, let’s create a SQLite database and define a simple schema for a blog post:

from sqlalchemy import create_engine, Column, Integer, String, Text
from sqlalchemy.orm import declarative_base

engine = create_engine('sqlite:///blog.db', echo=True)
Base = declarative_base()

class BlogPost(Base):
    __tablename__ = 'blog_post'

    id = Column(Integer, primary_key=True)
    title = Column(String(255))
    body = Column(Text)

In the above code, we are creating a SQLite database with the name blog.db and defining a table blog_post with three columns: id, title, and body. We also define a Base class that will be used to create the database schema.

Next, we need to create the database schema by running the following command:

Base.metadata.create_all(bind=engine)

Now, let’s define our FastAPI app and configure it to work with SQLAlchemy ORM:

from fastapi import FastAPI, HTTPException
from sqlalchemy.orm import Session

app = FastAPI()

@app.middleware("http")
async def db_session_middleware(request, call_next):
    response = None
    try:
        request.state.db = Session(bind=engine)
        response = await call_next(request)
    finally:
        request.state.db.close()
    return response

In the above code, we are defining a middleware function that will create a new database session for each request and close it after the request is processed. We use the Session class from SQLAlchemy to create the database session.

Next, let’s define our API endpoints for creating and fetching blog posts:

from pydantic import BaseModel

class BlogPostCreateRequest(BaseModel):
    title: str
    body: str

class BlogPostResponse(BaseModel):
    id: int
    title: str
    body: str

@app.post('/blog_posts')
def create_blog_post(request: BlogPostCreateRequest, db: Session = None):
    blog_post = BlogPost(title=request.title, body=request.body)
    db.add(blog_post)
    db.commit()
    db.refresh(blog_post)
    return BlogPostResponse(id=blog_post.id, title=blog_post.title, body=blog_post.body)

@app.get('/blog_posts/{blog_post_id}')
def get_blog_post(blog_post_id: int, db: Session = None):
    blog_post = db.query(BlogPost).filter(BlogPost.id == blog_post_id).first()
    if not blog_post:
        raise HTTPException(status_code=404, detail='Blog post not found')
    return BlogPostResponse(id=blog_post.id, title=blog_post.title, body=blog_post.body)

In the above code, we define two endpoints – create_blog_post for creating a new blog post and get_blog_post for fetching a blog post by ID. We use Pydantic models for input validation and response serialization.

To test our API, we can use a tool like httpie:

$ http POST :8000/blog_posts title="My First Blog Post" body="Hello, world!"
HTTP/1.1 200 OK
content-length