Building Pastebin clone using FastAPI

Pastebin is a website that allows you to share text data using a unique link. This comes in handy when you need to share a quick code snippet or any small amount of information. In this article, you'll see how to implement the text sharing using FastAPI, a very young and widely adopted asynchronous python framework.

Pastebin offers a lot of features while sharing text data. We'll just be implementing two.

Sharing text data using a unique link
Creating passwords for protecting the data(optional)

Let's get straight into business.

Install the dependencies.

pip install fastapi==0.63.0 uvicorn==0.13.3 SQLAlchemy==1.3.22 databases==0.4.1 aiosqlite==0.16.0 passlib==1.7.4 bcrypt==3.2.0

Let's see what each of them does.

Library	Description
fastapi	The core of our application
uvicorn	ASGI server for fastapi
sqlalchemy	To construct database models and SQL queries
databases	Gives you simple asyncio support for a range of databases.
aiosqlite	async interface to SQLite database
passlib	to create password hashes

The project structure,

.
├── app
│   ├── __init__.py
│   ├── api.py
│   ├── database.py
│   └── schemas.py
├── main.py
└── requirements.txt

Define the schemas

# app/schemas.py

from typing import Optional

from pydantic import BaseModel, Field


class BaseField(BaseModel):
    password: Optional[str]


class AddPaste(BaseField):
    text: str = Field(...)


class GetPaste(BaseField):
    id: str = Field(...)

To add a new paste(yeah, that's what we call it now), you need text data(string) and a password(optional). To retrieve a paste, you need the unique identifier and password(if the paste requires one).

Define the database models

# app/database.py

import databases

import sqlalchemy
from sqlalchemy import Column, String

DB_URI = "sqlite:///./test.db"

database = databases.Database(DB_URI)

metadata = sqlalchemy.MetaData()

paste = sqlalchemy.Table(
    "paste",
    metadata,
    Column("id", String, primary_key=True),
    Column("text", String, nullable=False),
    Column("password_hash", String, nullable=True),
)

engine = sqlalchemy.create_engine(
    DB_URI,
    connect_args={"check_same_thread": False},
)

metadata.create_all(engine)

The paste table will store a paste. It contains three fields,

id: a unique identifier
text: the text data to store
password_hash: hash of the password provided

Define the endpoints

Create a new router and a helper function to create unique ids.

# app/api.py

from fastapi import APIRouter

api = APIRouter(prefix="/api")

def generate_uid() -> str:
    return uuid.uuid4().hex[:6]

First, we define an endpoint to add a new paste.

# app/api.py

from passlib.hash import bcrypt
from app.database import database, paste
from app.schemas import AddPaste, GetPaste

@api.post("/")
async def add_paste(req: AddPaste):
    id = generate_uid()
    if req.password:
        req.password = bcrypt.hash(req.password)
        query = paste.insert().values(
            id=id,
            text=req.text,
            password_hash=req.password,
        )
    else:
        query = paste.insert().values(
            id=id,
            text=req.text,
        )
    _ = await database.execute(query)
    return id

We assume the collision rate of unique ids generated is zero. This assumption is evil, but we'll manage for now.

We hash the password using bcrypt and store the hashed password.

Now, we define the endpoint to retrieve a paste.

# app/api.py

from fastapi import HTTPException, status

@api.post("/get")
async def retrieve_paste(req: GetPaste):
    query = paste.select().where(paste.c.id == req.id)
    res = await database.fetch_one(query)
    if res is None:
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST,
            detail="invalid id",
        )
    if res.password_hash is not None:
        if req.password == "" or req.password is None:
            raise HTTPException(
                status_code=status.HTTP_403_FORBIDDEN,
                detail="Password required",
            )
        if bcrypt.verify(req.password, res.password_hash):
            return res.text
        else:
            raise HTTPException(
                status_code=status.HTTP_401_UNAUTHORIZED,
                detail="Wrong password",
            )
    return res.text

The code is self-explanatory.

We raise a couple of exceptions here.

If the unique code does not exist(invalid)
If the password is required and not provided
If the password provided is wrong

Finally, we need to create an app and include api in our app.

# main.py

from fastapi import FastAPI

from app.api import api
from app.database import database

app = FastAPI()
app.include_router(api)

Use FastAPI Events to create the database connection,

# main.py

@app.on_event("startup")
async def startup():
    await database.connect()

@app.on_event("shutdown")
async def shutdown():
    await database.disconnect()

The startup event runs before the app starts. It uses the databases library to create and manage the connection pool. We close the connections using the shutdown event.

Fire up the app by running uvicorn main:app and test out using interactive docs at localhost:8000/docs