Build a Video Membership Web App with Python, NoSQL, & FastAPI from the Ground Up

In this tutorial, I will guide you through the process of creating a Video Membership Web App from scratch using Python, NoSQL, and FastAPI. FastAPI is a modern, fast (high-performance), web framework for building APIs with Python. NoSQL databases are becoming increasingly popular for building web applications due to their flexibility and scalability.

We will be using MongoDB as our NoSQL database. MongoDB is a document-oriented NoSQL database that stores data in flexible, JSON-like documents. It is highly scalable and can handle large amounts of data efficiently.

By the end of this tutorial, you will have a fully functional Video Membership Web App where users can sign up, log in, view and upload videos, and subscribe to different membership plans.

Prerequisites:

• Basic understanding of Python
• Familiarity with API development concepts
• Basic knowledge of MongoDB
• Install Python on your machine
• Install MongoDB on your machine
• Install FastAPI using pip

Let’s get started!

Step 1: Setting up the project
Create a new directory for your project and navigate to it in the terminal. Create a new Python virtual environment using the following command:

python -m venv venv

Activate the virtual environment using:

• For Windows: venvScriptsactivate
• For macOS/Linux: source venv/bin/activate

Install FastAPI, Pydantic, and Motor using pip:

pip install fastapi uvicorn pydantic motor

Step 2: Setting up MongoDB
Start MongoDB on your machine using the following command:

mongod

Create a new database for your project using the MongoDB shell:

mongo
use video_membership
exit

Step 3: Create a FastAPI app
Create a new Python file named main.py in your project directory and add the following code:

from fastapi import FastAPI

app = FastAPI()

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

Run the FastAPI app using Uvicorn:

uvicorn main:app --reload

Navigate to http://localhost:8000 in your browser to see the "Hello World" message.

Step 4: Create models for users and videos
Create a new Python file named models.py in your project directory and add the following code:

from pydantic import BaseModel
from bson import ObjectId

class User(BaseModel):
    id: ObjectId
    username: str
    email: str
    password: str

class Video(BaseModel):
    id: ObjectId
    title: str
    description: str
    url: str

Step 5: Create CRUD endpoints for users and videos
Update the main.py file with the following code to create CRUD endpoints for users and videos:

from fastapi import FastAPI, HTTPException
from typing import List
from models import User, Video
from motor.motor_asyncio import AsyncIOMotorClient

app = FastAPI()
client = AsyncIOMotorClient()
db = client.video_membership

@app.post("/users/", response_model=User)
async def create_user(user: User):
    result = await db.users.insert_one(user.dict(exclude_unset=True))
    user.id = result.inserted_id
    return user

@app.get("/users/", response_model=List[User])
async def get_users():
    users = await db.users.find().to_list(1000)
    return users

@app.post("/videos/", response_model=Video)
async def create_video(video: Video):
    result = await db.videos.insert_one(video.dict(exclude_unset=True))
    video.id = result.inserted_id
    return video

@app.get("/videos/", response_model=List[Video])
async def get_videos():
    videos = await db.videos.find().to_list(1000)
    return videos

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8000)

Step 6: Testing the CRUD endpoints
Run the FastAPI app using Uvicorn and navigate to http://localhost:8000/docs in your browser to see the Swagger UI documentation. You can test the CRUD endpoints for users and videos from the Swagger UI.

Step 7: Implement user authentication
Update the main.py file with the following code to implement user authentication using JSON Web Tokens (JWT):

from fastapi import FastAPI, HTTPException, Depends
from fastapi.security import OAuth2PasswordBearer
from jwt import encode, decode, ExpiredSignatureError
from datetime import datetime, timedelta
from passlib.context import CryptContext
from models import User, Video
from motor.motor_asyncio import AsyncIOMotorClient

app = FastAPI()
client = AsyncIOMotorClient()
db = client.video_membership

SECRET_KEY = "supersecretkey"
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 30

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/token")
pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")

def create_access_token(data: dict):
    expire = datetime.utcnow() + timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
    to_encode = data.copy()
    to_encode.update({"exp": expire})
    return encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)

async def get_current_user(token: str = Depends(oauth2_scheme)):
    try:
        payload = decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        username = payload["sub"]
        user = await db.users.find_one({"username": username})
        if user is None:
            raise HTTPException(status_code=401, detail="Invalid credentials")
        return user
    except ExpiredSignatureError:
        raise HTTPException(status_code=401, detail="Token has expired")

@app.post("/token")
async def login_for_access_token(user: User):
    user = await db.users.find_one({"username": user.username})
    if user is None or not pwd_context.verify(user.password, user.password):
        raise HTTPException(status_code=401, detail="Invalid username or password")
    return {"access_token": create_access_token({"sub": user.username})}

@app.post("/users/", response_model=User)
async def create_user(user: User):
    hashed_password = pwd_context.hash(user.password)
    user.password = hashed_password
    result = await db.users.insert_one(user.dict(exclude_unset=True))
    user.id = result.inserted_id
    return user

Step 8: Add video upload and subscription endpoints
Update the main.py file with the following code to add video upload and subscription endpoints:

@app.post("/videos/", response_model=Video)
async def create_video(video: Video, current_user: User = Depends(get_current_user)):
    result = await db.videos.insert_one(video.dict(exclude_unset=True))
    video.id = result.inserted_id
    return video

@app.post("/subscribe/")
async def subscribe_to_membership(user: User, current_user: User = Depends(get_current_user)):
    # implement subscription logic here
    pass

Step 9: Testing the Video Membership Web App
Run the FastAPI app using Uvicorn and navigate to http://localhost:8000/docs in your browser to test the Video Membership Web App.

Congratulations! You have successfully created a Video Membership Web App from scratch using Python, NoSQL, and FastAPI. You can further enhance the app by adding features like video playback, user profiles, payment integration, and more. Feel free to explore FastAPI’s documentation and MongoDB’s documentation for more advanced features and capabilities. Happy coding!