Building Pastebin clone using FastAPI

Featured on Hashnode

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.

LibraryDescription
fastapiThe core of our application
uvicornASGI server for fastapi
sqlalchemyTo construct database models and SQL queries
databasesGives you simple asyncio support for a range of databases.
aiosqliteasync interface to SQLite database
passlibto 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

Results

Create a new paste

image.png

Retrieve a paste

image.png

If you liked this post, go ahead and try more features by yourself.

No Comments Yet