Webtops

Prompt 2 – Build a simple rest api using python’s fastapi for a todo app. include the following: jwt based authentication with login and signup endpoints. endpoints to add, update, delete, and list todos (crud). protect crud endpoints so only logged in users can access them

Written by

aghahasaan

in

Prompt # 2

Build a simple rest api using python’s fastapi for a todo app. include the following: jwt based authentication with login and signup endpoints. endpoints to add, update, delete, and list todos (crud). protect crud endpoints so only logged in users can access them.

Evaluation by Gemini Flash 2.0

Model:

Gemini Flash 2.0 (Experimental)

Gemini Pro

Rating: 4 Grade: B+ Critique: Provides a well-structured FastAPI example with clear separation of concerns into modules (crud, models, schemas, etc.) which is suitable for larger projects. The code includes error handling with HTTPException, uses type hinting for readability, and has good commenting. It correctly uses status.HTTP_204_NO_CONTENT for delete operations. However, it lacks specific details on user authentication implementation (security.py) which might require a separate prompt to make the provided code fully runnable.

Gemini Flash

Rating: 4 Grade: B Critique: The response includes a complete example with in-memory database handling using databases and aiosqlite. It has a good Pydantic models setup for data validation. JWT authentication is implemented directly and the code is more self-contained compared to Gemini Pro, thus is easier to run directly. Lacks specific separation of concern to different files, and the code is all in one file.

Gemma 2 (9B)

Rating: 3.2 Grade: C Critique: The example provides a basic implementation with in-memory data and authentication. It’s the simplest approach compared to the others, and while functional for demonstration purposes, it lacks any kind of persistance. The code could have been more elaborate and it also doesn’t include any files. The explanations are helpful, but it does not reflect a production-ready setup.

Qwen Coder

Rating: 4 Grade: B Critique: Provides a relatively simple, one file implementation with comments to help understanding. It includes basic Pydantic models, JWT authentication and uses in-memory storage for user data. This example offers a good starting point for a simple app but has very basic data management and error handling.

LLaMa 3.1 (8b)

Rating: 3.5 Grade: C+ Critique: The example provides a structured approach using passlib for password hashing and jwt for token handling, including a token verification dependency function. It separates models and authentication logic but requires a real database for full functionality. Includes the requirements.txt which is useful. Includes some comments.

GPT 4o

Rating: 4.5 Grade: A- Critique: Offers a well-structured, modular setup with main.py, auth.py, models.py, and database.py, which are very good for larger applications. Uses SQLAlchemy with clear explanations for each step (including the structure of each file), it is one of the most comprehensive solutions. It has good error handling and the provided explanation section is very detailed. However, it may be harder to copy/paste and use directly because of the files separation. Missing a requirements.txt file.

01 mini

Rating: 4.8 Grade: A Critique: The response has excellent detailed code structure, with a single file for all of the functionality, which makes it immediately runnable. The code includes all necessary parts: database setup using SQLAlchemy, models, JWT authentication, and CRUD operations. The explanations are quite detailed and complete and the code is relatively easy to understand. The security considerations are well highlighted.

Gemini Pro 1.5

Prompt:

Build a simple rest api using python’s fastapi for a todo app. include the following: jwt based authentication with login and signup endpoints. endpoints to add, update, delete, and list todos (crud). protect crud endpoints so only logged in users can access them.

Model:

Gemini Pro

from fastapi import FastAPI, Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
from sqlalchemy.orm import Session
from datetime import timedelta

from . import crud, models, schemas
from .database import SessionLocal, engine

models.Base.metadata.create_all(bind=engine)

app = FastAPI()

# Dependency
def get_db():
db = SessionLocal()
try:
yield db
finally:
db.close()

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")

# --- Authentication ---

from .security import authenticate_user, create_access_token, ACCESS_TOKEN_EXPIRE_MINUTES

@app.post("/token", response_model=schemas.Token)
async def login_for_access_token(form_data: OAuth2PasswordRequestForm = Depends(), db: Session = Depends(get_db)):
user = authenticate_user(db, form_data.username, form_data.password)
if not user:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Incorrect username or password",
headers={"WWW-Authenticate": "Bearer"},
)
access_token_expires = timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
access_token = create_access_token(
data={"sub": user.username}, expires_delta=access_token_expires
)
return {"access_token": access_token, "token_type": "bearer"}

@app.post("/signup", response_model=schemas.User)
def create_user(user: schemas.UserCreate, db: Session = Depends(get_db)):
db_user = crud.get_user_by_username(db, username=user.username)
if db_user:
raise HTTPException(status_code=400, detail="Username already registered")
return crud.create_user(db=db, user=user)

# --- Todo CRUD ---
from .security import get_current_user

@app.post("/todos/", response_model=schemas.Todo)
def create_todo(todo: schemas.TodoCreate, db: Session = Depends(get_db), current_user: schemas.User = Depends(get_current_user)):
return crud.create_user_todo(db=db, todo=todo, user_id=current_user.id)

@app.get("/todos/", response_model=list[schemas.Todo])
def read_todos(skip: int = 0, limit: int = 100, db: Session = Depends(get_db), current_user: schemas.User = Depends(get_current_user)):
todos = crud.get_todos(db, skip=skip, limit=limit, user_id=current_user.id)
return todos

@app.get("/todos/{todo_id}", response_model=schemas.Todo)
def read_todo(todo_id: int, db: Session = Depends(get_db), current_user: schemas.User = Depends(get_current_user)):
db_todo = crud.get_todo(db, todo_id=todo_id, user_id=current_user.id)
if db_todo is None:
raise HTTPException(status_code=404, detail="Todo not found")
return db_todo

@app.put("/todos/{todo_id}", response_model=schemas.Todo)
def update_todo(todo_id: int, todo: schemas.TodoUpdate, db: Session = Depends(get_db), current_user: schemas.User = Depends(get_current_user)):
db_todo = crud.update_todo(db, todo_id=todo_id, todo=todo, user_id=current_user.id)
if db_todo is None:
raise HTTPException(status_code=404, detail="Todo not found")
return db_todo

@app.delete("/todos/{todo_id}", status_code=status.HTTP_204_NO_CONTENT)
def delete_todo(todo_id: int, db: Session = Depends(get_db), current_user: schemas.User = Depends(get_current_user)):
crud.delete_todo(db, todo_id=todo_id, user_id=current_user.id)
return None # No content

# For testing only. Uncomment for easier testing without a frontend.
# @app.get("/users/me/", response_model=schemas.User)
# async def read_users_me(current_user: schemas.User = Depends(get_current_user)):
# return current_user

This improved example includes:

  • Clearer Folder Structure: Assumes project structure with files like main.py, crud.py, models.py, schemas.py, database.py, and security.py. This organization is crucial for larger projects.
  • Error Handling: Includes HTTPException for better error responses.
  • Type Hinting: Uses type hints for better code readability and maintainability.
  • Concise Comments: More explanatory comments to understand the code’s purpose.
  • No Content Response: Correctly uses status.HTTP_204_NO_CONTENT for delete operations.

Supporting Files:

Create the following files in the same directory as main.py:

  • crud.py: Contains the database operations (create, read, update, delete).
  • models.py: Defines the database models (User, Todo).
  • schemas.py: Defines the Pydantic schemas for data validation.
  • database.py: Sets up the database connection.
  • security.py: Handles JWT token creation and user authentication.

Please let me know if you have another question.

Gemini Flash 1.5

Prompt:

Build a simple rest api using python’s fastapi for a todo app. include the following: jwt-based authentication with login and signup endpoints. endpoints to add, update, delete, and list todos (crud). protect crud endpoints so only logged-in users can access them.

Model:

Gemini Flash

This example uses FastAPI, uvicorn, PyJWT, and databases. You’ll need to install them: pip install fastapi uvicorn PyJWT databases aiosqlite (or replace aiosqlite with your preferred database driver like asyncpg for PostgreSQL).

import uvicorn
from fastapi import Depends, FastAPI, HTTPException, status
from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
from jose import JWTError, jwt
from pydantic import BaseModel
from datetime import datetime, timedelta
from databases import Database
import secrets

# Database setup (replace with your actual database connection string)
DATABASE_URL = "sqlite:///./todo.db" # Use SQLite for simplicity
database = Database(DATABASE_URL)

# JWT settings
SECRET_KEY = secrets.token_urlsafe(32) # Generate a strong secret key! DO NOT HARDCODE IN PRODUCTION
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 30

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")

# Pydantic models
class User(BaseModel):
id: int
username: str
password: str # Store password securely in real-world app (e.g., hashing)

class Todo(BaseModel):
id: int | None = None
task: str
completed: bool = False

class Token(BaseModel):
access_token: str
token_type: str

async def get_current_user(token: str = Depends(oauth2_scheme)):
credentials_exception = HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Could not validate credentials",
headers={"WWW-Authenticate": "Bearer"},
)
try:
payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
username: str = payload.get("sub")
if username is None:
raise credentials_exception
token_data = User(username=username, id=-1, password="") #Placeholder ID, only username is relevant here.
user = await get_user_by_username(token_data.username)
if user is None:
raise credentials_exception
return user
except JWTError:
raise credentials_exception

async def get_user_by_username(username: str):
query = "SELECT * FROM users WHERE username = :username"
user = await database.fetch_one(query, values={"username": username})
return User(**user) if user else None

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

app = FastAPI()

# Database migrations (run once to create the tables)
@app.on_event("startup")
async def startup():
await database.connect()
await database.execute(
"""
CREATE TABLE IF NOT EXISTS users (
id INTEGER PRIMARY KEY AUTOINCREMENT,
username TEXT UNIQUE NOT NULL,
password TEXT NOT NULL
)
"""
)
await database.execute(
"""
CREATE TABLE IF NOT EXISTS todos (
id INTEGER PRIMARY KEY AUTOINCREMENT,
task TEXT NOT NULL,
completed BOOLEAN NOT NULL,
user_id INTEGER NOT NULL,
FOREIGN KEY (user_id) REFERENCES users(id)
)
"""
)

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

# Authentication endpoints
@app.post("/token", response_model=Token)
async def login_for_access_token(form_data: OAuth2PasswordRequestForm = Depends()):
user = await get_user_by_username(form_data.username)
if not user:
raise HTTPException(status_code=400, detail="Incorrect username or password")
if user.password != form_data.password: #Insecure - replace with proper password hashing in production.
raise HTTPException(status_code=400, detail="Incorrect username or password")
access_token = await create_access_token(data={"sub": user.username})
return {"access_token": access_token, "token_type": "bearer"}

@app.post("/signup", response_model=User, status_code=201)
async def signup(user: User):
query = "INSERT INTO users (username, password) VALUES (:username, :password)"
await database.execute(query, values={"username": user.username, "password": user.password})
return user

# Todo endpoints (protected)
@app.post("/todos/", response_model=Todo, status_code=201)
async def create_todo(todo: Todo, user: User = Depends(get_current_user)):
query = "INSERT INTO todos (task, completed, user_id) VALUES (:task, :completed, :user_id)"
values = {"task": todo.task, "completed": todo.completed, "user_id": user.id} #Replace user.id with proper user id retrieval
await database.execute(query, values=values)
return todo

@app.get("/todos/", response_model=list[Todo])
async def list_todos(user: User = Depends(get_current_user)):
query = "SELECT * FROM todos WHERE user_id = :user_id"
todos = await database.fetch_all(query, values={"user_id": user.id}) #Replace user.id with proper user id retrieval
return [Todo(**todo) for todo in todos]

@app.put("/todos/{todo_id}", response_model=Todo)
async def update_todo(todo_id: int, todo: Todo, user: User = Depends(get_current_user)):
query = "UPDATE todos SET task = :task, completed = :completed WHERE id = :id AND user_id = :user_id"
values = {"task": todo.task, "completed": todo.completed, "id": todo_id, "user_id": user.id} #Replace user.id with proper user id retrieval
await database.execute(query, values=values)
return todo

@app.delete("/todos/{todo_id}", status_code=204)
async def delete_todo(todo_id: int, user: User = Depends(get_current_user)):
query = "DELETE FROM todos WHERE id = :id AND user_id = :user_id"
values = {"id": todo_id, "user_id": user.id} #Replace user.id with proper user id retrieval
await database.execute(query, values=values)

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

Remember to replace "sqlite:///./todo.db" with your actual database connection string if you’re not using SQLite. Critically, replace the placeholder password handling with a secure hashing algorithm (like bcrypt or Argon2) in a production environment. The provided code is for illustrative purposes only and is not suitable for production without significant security improvements. Also, consider adding error handling and input validation for a more robust application. Finally, replace the placeholder user.id with the actual user ID retrieval from your database.

Gemma 2 (9b)

Prompt:

Build a simple rest api using python’s fastapi for a todo app. include the following: jwt-based authentication with login and signup endpoints. endpoints to add, update, delete, and list todos (crud). protect crud endpoints so only logged-in users can access them.

Model:

Gemma 2

This example demonstrates a basic implementation of a REST API with FastAPI, JWT authentication, and CRUD operations for a todo app.

Remember: This is a simplified example and does not include features like database persistence or error handling for production use. You should implement these for a real-world application.

from fastapi import FastAPI, Depends, HTTPException
from pydantic import BaseModel
import jwt
from datetime import datetime, timedelta

app = FastAPI()

# Configuration
SECRET_KEY = "YOUR_SECRET_KEY" # Replace with a strong secret key!
ALGORITHM = "HS256"

class User(BaseModel):
username: str
password: str

class Todo(BaseModel):
id: int
title: str
completed: bool

todos = [
{"id": 1, "title": "Buy groceries", "completed": False},
{"id": 2, "title": "Learn FastAPI", "completed": True}
]

# Simulated User Database (replace with a real database)
users = {
"user1": {"password": "password123"}
}

def get_current_user(token: str = Depends(get_token)):
try:
payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
username = payload.get("username")
if username is None:
raise HTTPException(status_code=401, detail="Invalid token")
return username
except jwt.ExpiredSignatureError:
raise HTTPException(status_code=401, detail="Token expired")
except jwt.DecodeError:
raise HTTPException(status_code=401, detail="Token invalid")

def get_token(request):
authorization = request.headers.get("Authorization")
if authorization is None or not authorization.startswith("Bearer"):
return None
return authorization.split(" ")[1]

# Authentication Endpoints

@app.post("/signup")
async def signup(user: User):
existing_user = users.get(user.username)
if existing_user:
raise HTTPException(status_code=400, detail="Username already exists")

users[user.username] = {"password": user.password}
return {"message": "User created successfully"}

@app.post("/login")
async def login(user: User):
if user.username not in users:
raise HTTPException(status_code=401, detail="Invalid credentials")
if users[user.username]["password"] != user.password:
raise HTTPException(status_code=401, detail="Invalid credentials")

token = jwt.encode(
{"username": user.username, "exp": datetime.utcnow() + timedelta(minutes=30)},
SECRET_KEY,
algorithm=ALGORITHM
)
return {"access_token": token}

# Protected CRUD Endpoints (require JWT authentication)

@app.get("/todos", response_model=list[Todo], dependencies=[Depends(get_current_user)])
async def read_todos(current_user: str):
return todos

@app.post("/todos", response_model=Todo, dependencies=[Depends(get_current_user)])
async def create_todo(title: str, current_user: str):
new_todo = {"id": len(todos) + 1, "title": title, "completed": False}
todos.append(new_todo)
return new_todo

@app.put("/todos/{todo_id}", response_model=Todo, dependencies=[Depends(get_current_user)])
async def update_todo(todo_id: int, title: str = None, completed: bool = None, current_user: str):
for todo in todos:
if todo["id"] == todo_id:
if title is not None:
todo["title"] = title
if completed is not None:
todo["completed"] = completed
return todo

raise HTTPException(status_code=404, detail="Todo not found")

@app.delete("/todos/{todo_id}", dependencies=[Depends(get_current_user)])
async def delete_todo(todo_id: int, current_user: str):
for todo in todos:
if todo["id"] == todo_id:
todos.remove(todo)
return {"message": "Todo deleted successfully"}
raise HTTPException(status_code=404, detail="Todo not found")

Explanation:

  1. Dependencies and Models:

    • We use FastAPI’s Depends to control access to endpoints based on JWT authentication.
    • BaseModel from Pydantic is used for data validation (e.g., checking if the “username” and “password” fields are present).

  2. Secret Key and Algorithm:

    • Replace "YOUR_SECRET_KEY" with a strong, unique secret key that should be kept secure. It’s crucial for JWT signature verification.
    • ALGORITHM = "HS256" specifies the hashing algorithm used for generating the JWT signatures.

  3. User Database (Simulated):

    • This example uses a simple dictionary (users) to store usernames and passwords. In a real app, you’d use a database.

  4. JWT Token Handling:

    • The get_token function extracts the authorization token from the request headers.
    • The jwt.decode function validates and decodes the JWT token.
    • If the token is valid, it returns the username; otherwise, an HTTPException (401 Unauthorized) is raised.

  5. Authentication Endpoints:

    • /signup: Creates a new user in the simulated database and stores their password securely (important to hash passwords properly!)
    • /login: Verifies user credentials and generates a JWT token containing the username and an expiration time (30 minutes in this example).

  6. Protected CRUD Endpoints:

    • All endpoints for creating, reading, updating, and deleting todos (/todos, /todos/{todo_id}, etc.) are protected by get_current_user. This function ensures that only authenticated users can access these resources.

How to Run:

  1. Install Dependencies: pip install fastapi uvicorn pydantic python-jose
  2. Run the Server: uvicorn main:app --reload (assuming your file is named main.py)

Let me know if you have any more questions or would like to explore additional features!

Qwen-Coder 2.5 (9b)

Prompt:

Build a simple rest api using python’s fastapi for a todo app. include the following: jwt-based authentication with login and signup endpoints. endpoints to add, update, delete, and list todos (crud). protect crud endpoints so only logged-in users can access them.

Model:

Qwen 2.5 Coder

Here’s an example of how you could build a simple REST API using Python’s FastAPI for a todo app with JWT-based authentication:

Firstly, install the necessary libraries by running:

pip install fastapi uvicorn pydantic python-jose[cryptography]

Next, create a file called main.py and add the following code:

from fastapi import Depends, FastAPI, HTTPException, status
from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
from jose import JWTError, jwt
from pydantic import BaseModel
from datetime import datetime, timedelta

# Set up security and database models (simplified)
SECRET_KEY = "your-secret-key"
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 30

fake_users_db = {
"johndoe": {
"username": "johndoe",
"full_name": "John Doe",
"email": "johndoe@example.com",
"hashed_password": "$2b$12$EixZaYVK1fsbw1ZfbX3OXePaWxn96p36WQoeG6Lruj3vjPGga31lW", # hashed password
"disabled": False,
}
}

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")

app = FastAPI()

class Token(BaseModel):
access_token: str
token_type: str

class TokenData(BaseModel):
username: str | None = None

class User(BaseModel):
username: str
email: str | None = None
full_name: str | None = None
disabled: bool | None = None

class UserInDB(User):
hashed_password: str

def get_user(db, username: str):
if username in db:
user_dict = db[username]
return UserInDB(**user_dict)

def authenticate_user(fake_db, username: str, password: str):
user = get_user(fake_db, username)
if not user:
return False
if not verify_password(password, user.hashed_password):
return False
return user

def create_access_token(data: dict, expires_delta: timedelta | None = None):
to_encode = data.copy()
if expires_delta:
expire = datetime.utcnow() + expires_delta
else:
expire = datetime.utcnow() + timedelta(minutes=15)
to_encode.update({"exp": expire})
encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
return encoded_jwt

async def get_current_user(token: str = Depends(oauth2_scheme)):
credentials_exception = HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Could not validate credentials",
headers={"WWW-Authenticate": "Bearer"},
)
try:
payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
username: str = payload.get("sub")
if username is None:
raise credentials_exception
token_data = TokenData(username=username)
except JWTError:
raise credentials_exception
user = get_user(fake_users_db, username=token_data.username)
if user is None:
raise credentials_exception
return user

@app.post("/token", response_model=Token)
async def login_for_access_token(form_data: OAuth2PasswordRequestForm = Depends()):
user = authenticate_user(fake_users_db, form_data.username, form_data.password)
if not user:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Incorrect username or password",
headers={"WWW-Authenticate": "Bearer"},
)
access_token_expires = timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
access_token = create_access_token(
data={"sub": user.username}, expires_delta=access_token_expires
)
return {"access_token": access_token, "token_type": "bearer"}

# Protected CRUD endpoints (simplified)
@app.get("/todos")
async def read_todos(current_user: User = Depends(get_current_user)):
# Logic to fetch todos for the current user
pass

@app.post("/todos", response_model=User)
async def create_todo(todo: User, current_user: User = Depends(get_current_user)):
# Logic to create a new todo for the current user
pass

@app.put("/todos/{todo_id}", response_model=User)
async def update_todo(todo_id: int, todo: User, current_user: User = Depends(get_current_user)):
# Logic to update an existing todo for the current user
pass

@app.delete("/todos/{todo_id}", response_model=User)
async def delete_todo(todo_id: int, current_user: User = Depends(get_current_user)):
# Logic to delete a todo for the current user
pass

This is a simplified version of how you could implement a REST API with FastAPI, JWT-based authentication, and protected CRUD endpoints. Make sure to replace "your-secret-key" with a real secret key and customize the database models and logic as needed.

LLaMA 3.1 (8.1)

Prompt:

Build a simple rest api using python’s fastapi for a todo app. include the following: jwt-based authentication with login and signup endpoints. endpoints to add, update, delete, and list todos (crud). protect crud endpoints so only logged-in users can access them.

Model:

LLaMA 3.1

Here’s an example of a simple REST API using Python’s FastAPI for a todo app:

main.py

from fastapi import FastAPI, Depends
from pydantic import BaseModel
import jwt
from passlib.context import CryptContext

# Define a secret key for JWT
SECRET_KEY = "your_secret_key_here"

# Initialize the app
app = FastAPI()

# Define a crypt context for password hashing
pwd_context = CryptContext(schemes=["bcrypt"], default="bcrypt")

# Define models for User and Todo
class User(BaseModel):
username: str
email: str
password: str

class Todo(BaseModel):
id: int
title: str
completed: bool

# Define a dependency function to authenticate users
def authenticate_user(username: str, password: str):
# In a real application, this would be a database query
return User(username="john", email="john@example.com", password=pwd_context.hash(password))

# Define a token verification function
def verify_token(token: str = Depends()):
try:
payload = jwt.decode(token, SECRET_KEY, algorithms=["HS256"])
except jwt.ExpiredSignatureError:
raise HTTPException(status_code=401, detail="Token has expired")
except jwt.InvalidToken:
raise HTTPException(status_code=401, detail="Invalid token")

# Define a login endpoint
@app.post("/login")
async def login(form_data: User):
user = authenticate_user(form_data.username, form_data.password)
access_token = jwt.encode({"username": user.username}, SECRET_KEY, algorithm="HS256")
return {"access_token": access_token}

# Define a signup endpoint
@app.post("/signup")
async def signup(user_data: User):
hashed_password = pwd_context.hash(user_data.password)
db_user = User(username=user_data.username, email=user_data.email, password=hashed_password) # In a real application, this would be a database insertion
return {"message": "User created successfully"}

# Define CRUD endpoints for todos, protected by JWT authentication
@app.get("/todos")
async def get_todos(token: str = Depends(verify_token)):
# In a real application, this would be a database query
todos = [Todo(id=1, title="Buy milk", completed=False), Todo(id=2, title="Walk dog", completed=True)]
return {"todos": todos}

@app.post("/todos")
async def create_todo(todo_data: Todo, token: str = Depends(verify_token)):
# In a real application, this would be a database insertion
return {"message": "Todo created successfully"}

@app.put("/todos/{todo_id}")
async def update_todo(todo_id: int, todo_data: Todo, token: str = Depends(verify_token)):
# In a real application, this would be a database query and update
if todo_id == 1:
return {"message": "Todo updated successfully"}
else:
raise HTTPException(status_code=404, detail="Todo not found")

@app.delete("/todos/{todo_id}")
async def delete_todo(todo_id: int, token: str = Depends(verify_token)):
# In a real application, this would be a database deletion
return {"message": "Todo deleted successfully"}

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

requirements.txt

fastapi
pydantic
jwt
passlib
uvicorn

This code defines a simple REST API for a todo app with JWT-based authentication and CRUD endpoints for todos. The login endpoint generates an access token using the authenticate_user function, which is then used to protect the CRUD endpoints.

Note that this is just an example and in a real application you should use a secure secret key, a database instead of mock data, and implement proper error handling and logging.

To run this code:

  1. Create a new virtual environment and activate it.
  2. Install the required packages using pip: pip install -r requirements.txt
  3. Run the app with uvicorn: uvicorn main:app --host 0.0.0.0 --port 8000

You can now access the API endpoints at http://localhost:8000.

Example use cases:

  • Use a tool like Postman or cURL to send requests to the API endpoints.
  • Use a web browser to access the /login and /signup endpoints and test the JWT authentication flow.
  • Use a tool like insomnia to access the CRUD endpoints for todos.

GPT-4o

Prompt:

Build a simple rest api using python’s fastapi for a todo app. include the following: jwt-based authentication with login and signup endpoints. endpoints to add, update, delete, and list todos (crud). protect crud endpoints so only logged-in users can access them.

Model:

GPT 4o

Creating a simple REST API for a todo app using FastAPI involves several steps, including setting up JWT authentication for user login and signup, and implementing the CRUD operations for todos. Below is an example of how you can achieve this:

First, make sure you have FastAPI, uvicorn, and pyjwt installed. You can install them using pip if you haven’t already:

pip install fastapi uvicorn python-jose bcrypt pydantic

Here’s a basic implementation:

Directory Structure

todo_app/
├── main.py
├── auth.py
├── models.py
└── database.py

main.py

This is the main entry point of your FastAPI application.

from fastapi import FastAPI, Depends, HTTPException
from fastapi.security import OAuth2PasswordRequestForm
from sqlalchemy.orm import Session
from . import models, auth, database
from .auth import get_current_user

app = FastAPI()

# Create the database tables
models.Base.metadata.create_all(bind=database.engine)

@app.post('/signup')
def signup(user: models.UserCreate, db: Session = Depends(database.get_db)):
db_user = auth.create_user(db, user)
return {"message": "User created successfully", "username": db_user.username}

@app.post('/login')
def login(form_data: OAuth2PasswordRequestForm = Depends(), db: Session = Depends(database.get_db)):
user = auth.authenticate_user(db, form_data.username, form_data.password)
if not user:
raise HTTPException(status_code=400, detail="Invalid credentials")
access_token = auth.create_access_token(data={"sub": user.username})
return {"access_token": access_token, "token_type": "bearer"}

@app.post('/todos', response_model=models.Todo)
def create_todo(todo: models.TodoCreate, db: Session = Depends(database.get_db), current_user: models.User = Depends(get_current_user)):
return auth.create_user_todo(db, todo, current_user)

@app.get('/todos', response_model=list[models.Todo])
def get_todos(db: Session = Depends(database.get_db), current_user: models.User = Depends(get_current_user)):
return auth.get_user_todos(db, current_user)

@app.put('/todos/{todo_id}', response_model=models.Todo)
def update_todo(todo_id: int, todo: models.TodoCreate, db: Session = Depends(database.get_db), current_user: models.User = Depends(get_current_user)):
return auth.update_user_todo(db, todo_id, todo, current_user)

@app.delete('/todos/{todo_id}')
def delete_todo(todo_id: int, db: Session = Depends(database.get_db), current_user: models.User = Depends(get_current_user)):
return auth.delete_user_todo(db, todo_id, current_user)

auth.py

This file manages authentication and user management.

from fastapi import HTTPException, Depends
from fastapi.security import OAuth2PasswordBearer
from jose import JWTError, jwt
from sqlalchemy.orm import Session
from . import models, database
from passlib.context import CryptContext
from datetime import datetime, timedelta

# Fake the SECRET_KEY, ALGORITHM, and ACCESS_TOKEN_EXPIRE_MINUTES
SECRET_KEY = "SECRET"
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 30

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

def create_user(db: Session, user: models.UserCreate):
hashed_password = pwd_context.hash(user.password)
db_user = models.User(username=user.username, hashed_password=hashed_password)
db.add(db_user)
db.commit()
db.refresh(db_user)
return db_user

def authenticate_user(db: Session, username: str, password: str):
user = db.query(models.User).filter(models.User.username == username).first()
if not user or not pwd_context.verify(password, user.hashed_password):
return False
return user

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

def get_current_user(db: Session = Depends(database.get_db), token: str = Depends(oauth2_scheme)):
credentials_exception = HTTPException(
status_code=401,
detail="Could not validate credentials",
headers={"WWW-Authenticate": "Bearer"},
)
try:
payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
username: str = payload.get("sub")
if username is None:
raise credentials_exception
except JWTError:
raise credentials_exception
user = db.query(models.User).filter(models.User.username == username).first()
if user is None:
raise credentials_exception
return user

def create_user_todo(db: Session, todo: models.TodoCreate, current_user: models.User):
db_todo = models.Todo(**todo.dict(), owner_id=current_user.id)
db.add(db_todo)
db.commit()
db.refresh(db_todo)
return db_todo

def get_user_todos(db: Session, current_user: models.User):
return db.query(models.Todo).filter(models.Todo.owner_id == current_user.id).all()

def update_user_todo(db: Session, todo_id: int, todo: models.TodoCreate, current_user: models.User):
db_todo = db.query(models.Todo).filter(models.Todo.id == todo_id, models.Todo.owner_id == current_user.id).first()
if not db_todo:
raise HTTPException(status_code=404, detail="Todo not found")
for key, value in todo.dict().items():
setattr(db_todo, key, value)
db.commit()
db.refresh(db_todo)
return db_todo

def delete_user_todo(db: Session, todo_id: int, current_user: models.User):
db_todo = db.query(models.Todo).filter(models.Todo.id == todo_id, models.Todo.owner_id == current_user.id).first()
if not db_todo:
raise HTTPException(status_code=404, detail="Todo not found")
db.delete(db_todo)
db.commit()
return {"detail": "Todo deleted"}

models.py

This file defines the database models using SQLAlchemy.

from sqlalchemy import Column, Integer, String, ForeignKey
from sqlalchemy.orm import relationship
from pydantic import BaseModel
from .database import Base

class User(Base):
__tablename__ = 'users'
id = Column(Integer, primary_key=True, index=True)
username = Column(String, unique=True, index=True)
hashed_password = Column(String)
todos = relationship('Todo', back_populates='owner')

class Todo(Base):
__tablename__ = 'todos'
id = Column(Integer, primary_key=True, index=True)
title = Column(String, index=True)
description = Column(String, index=True)
owner_id = Column(Integer, ForeignKey('users.id'))
owner = relationship('User', back_populates='todos')

class UserCreate(BaseModel):
username: str
password: str

class TodoBase(BaseModel):
title: str
description: str

class TodoCreate(TodoBase):
pass

class Todo(TodoBase):
id: int
owner_id: int

class Config:
orm_mode = True

database.py

This file manages the database connection using SQLAlchemy.

from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker

SQLALCHEMY_DATABASE_URL = "sqlite:///./test.db" # Use SQLite for simplicity

engine = create_engine(
SQLALCHEMY_DATABASE_URL, connect_args={"check_same_thread": False}
)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
Base = declarative_base()

def get_db():
db = SessionLocal()
try:
yield db
finally:
db.close()

Running the Application

To run the application, use the following command:

uvicorn todo_app.main:app --reload

Note

  • This example uses a simple SQLite database.
  • Passwords are hashed using passlib for security.
  • In a real-world application, you’d want to handle various exceptions more robustly and increase the complexity of hashing and salting strategies.
  • This example follows the OAuth2 password flow for obtaining JWT tokens.
  • You should use environment variables for the SECRET_KEY and database URLs in a production environment.

This should provide you with a simple but functional REST API for a todo app using FastAPI, JWT authentication, and SQLAlchemy for database interactions.

o1-mini

Prompt:

Build a simple rest api using python’s fastapi for a todo app. include the following: jwt-based authentication with login and signup endpoints. endpoints to add, update, delete, and list todos (crud). protect crud endpoints so only logged-in users can access them.

Model:

o1-mini

Sure! Below is a complete example of a simple REST API for a Todo application using Python’s FastAPI framework. This API includes:

  • JWT-based Authentication: Users can sign up and log in to receive JWT tokens.
  • CRUD Endpoints: Authenticated users can create, read, update, and delete their todos.
  • Protection of Endpoints: Only logged-in users with valid JWT tokens can access the CRUD operations.

Prerequisites

First, ensure you have Python installed. Then, install the necessary dependencies using pip:

pip install fastapi uvicorn sqlalchemy passlib[bcrypt] python-jose

Complete Code Example

Here’s the complete code for the Todo API using FastAPI:

from fastapi import FastAPI, Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
from typing import List
from sqlalchemy import Column, Integer, String, ForeignKey, create_engine
from sqlalchemy.orm import sessionmaker, relationship, declarative_base, Session
from passlib.context import CryptContext
from jose import JWTError, jwt
from datetime import datetime, timedelta
from pydantic import BaseModel

# ======================= #
# CONFIG #
# ======================= #

SECRET_KEY = "your_secret_key_here" # Replace with your actual secret key
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 30

DATABASE_URL = "sqlite:///./test.db" # Using SQLite for simplicity

# ======================= #
# DATABASE SETUP #
# ======================= #

Base = declarative_base()

class User(Base):
__tablename__ = 'users'

id = Column(Integer, primary_key=True, index=True)
username = Column(String, unique=True, index=True)
hashed_password = Column(String)

todos = relationship("Todo", back_populates="owner")

class Todo(Base):
__tablename__ = 'todos'

id = Column(Integer, primary_key=True, index=True)
title = Column(String, index=True)
description = Column(String, nullable=True)
owner_id = Column(Integer, ForeignKey('users.id'))

owner = relationship("User", back_populates="todos")

# Create the database engine
engine = create_engine(
DATABASE_URL, connect_args={"check_same_thread": False}
)

# Create all tables
Base.metadata.create_all(bind=engine)

# Create a SessionLocal class
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

# ======================= #
# SECURITY SETUP #
# ======================= #

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

def verify_password(plain_password, hashed_password):
return pwd_context.verify(plain_password, hashed_password)

def get_password_hash(password):
return pwd_context.hash(password)

def create_access_token(data: dict, expires_delta: timedelta | None = None):
to_encode = data.copy()
if expires_delta:
expire = datetime.utcnow() + expires_delta
else:
expire = datetime.utcnow() + timedelta(minutes=15)
to_encode.update({"exp": expire})
encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
return encoded_jwt

# ======================= #
# MODELS #
# ======================= #

class UserCreate(BaseModel):
username: str
password: str

class Token(BaseModel):
access_token: str
token_type: str

class TodoCreate(BaseModel):
title: str
description: str | None = None

class TodoUpdate(BaseModel):
title: str | None = None
description: str | None = None

class TodoResponse(BaseModel):
id: int
title: str
description: str | None

class Config:
orm_mode = True

# ======================= #
# DEPENDENCIES #
# ======================= #

app = FastAPI()

def get_db():
db = SessionLocal()
try:
yield db
finally:
db.close()

def get_user(db: Session, username: str):
return db.query(User).filter(User.username == username).first()

def authenticate_user(db: Session, username: str, password: str):
user = get_user(db, username)
if not user:
return False
if not verify_password(password, user.hashed_password):
return False
return user

def get_current_user(token: str = Depends(oauth2_scheme), db: Session = Depends(get_db)):
credentials_exception = HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Could not validate credentials",
headers={"WWW-Authenticate": "Bearer"},
)
try:
payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
username: str = payload.get("sub")
if username is None:
raise credentials_exception
except JWTError:
raise credentials_exception
user = get_user(db, username=username)
if user is None:
raise credentials_exception
return user

# ======================= #
# ROUTES #
# ======================= #

@app.post("/signup", response_model=Token)
def signup(user_create: UserCreate, db: Session = Depends(get_db)):
user = get_user(db, username=user_create.username)
if user:
raise HTTPException(status_code=400, detail="Username already registered")
hashed_password = get_password_hash(user_create.password)
new_user = User(username=user_create.username, hashed_password=hashed_password)
db.add(new_user)
db.commit()
db.refresh(new_user)
access_token = create_access_token(
data={"sub": new_user.username},
expires_delta=timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
)
return {"access_token": access_token, "token_type": "bearer"}

@app.post("/login", response_model=Token)
def login(form_data: OAuth2PasswordRequestForm = Depends(), db: Session = Depends(get_db)):
user = authenticate_user(db, form_data.username, form_data.password)
if not user:
raise HTTPException(
status_code=401,
detail="Incorrect username or password",
headers={"WWW-Authenticate": "Bearer"},
)
access_token = create_access_token(
data={"sub": user.username},
expires_delta=timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
)
return {"access_token": access_token, "token_type": "bearer"}

@app.post("/todos/", response_model=TodoResponse)
def create_todo(todo: TodoCreate, current_user: User = Depends(get_current_user), db: Session = Depends(get_db)):
new_todo = Todo(title=todo.title, description=todo.description, owner_id=current_user.id)
db.add(new_todo)
db.commit()
db.refresh(new_todo)
return new_todo

@app.get("/todos/", response_model=List[TodoResponse])
def read_todos(skip: int = 0, limit: int = 10, current_user: User = Depends(get_current_user), db: Session = Depends(get_db)):
todos = db.query(Todo).filter(Todo.owner_id == current_user.id).offset(skip).limit(limit).all()
return todos

@app.put("/todos/{todo_id}", response_model=TodoResponse)
def update_todo(todo_id: int, todo: TodoUpdate, current_user: User = Depends(get_current_user), db: Session = Depends(get_db)):
db_todo = db.query(Todo).filter(Todo.id == todo_id, Todo.owner_id == current_user.id).first()
if not db_todo:
raise HTTPException(status_code=404, detail="Todo not found")
if todo.title is not None:
db_todo.title = todo.title
if todo.description is not None:
db_todo.description = todo.description
db.commit()
db.refresh(db_todo)
return db_todo

@app.delete("/todos/{todo_id}", status_code=204)
def delete_todo(todo_id: int, current_user: User = Depends(get_current_user), db: Session = Depends(get_db)):
db_todo = db.query(Todo).filter(Todo.id == todo_id, Todo.owner_id == current_user.id).first()
if not db_todo:
raise HTTPException(status_code=404, detail="Todo not found")
db.delete(db_todo)
db.commit()
return

# ======================= #
# RUNNING THE APP #
# ======================= #

# To run the application, use the following command:
# uvicorn your_file_name:app --reload

Explanation of the Code

  1. Configuration & Setup:

    • Secret Key & JWT Settings: Define SECRET_KEY, ALGORITHM, and token expiration time.
    • Database: Using SQLite for simplicity with SQLAlchemy ORM. Two models are defined: User and Todo. Each Todo is linked to a User.

  2. Security:

    • Password Hashing: Using passlib with bcrypt to hash and verify passwords.
    • JWT Token Creation: A function to create JWT tokens with an expiration time.
    • OAuth2 Integration: Using OAuth2PasswordBearer to handle token retrieval from requests.

  3. Models:

    • Pydantic Models: Define UserCreate, Token, TodoCreate, TodoUpdate, and TodoResponse for request validation and response formatting.

  4. Dependencies:

    • Database Session: get_db dependency provides a database session per request.
    • User Authentication: Functions to retrieve a user, authenticate a user, and get the current authenticated user based on the JWT token.

  5. API Routes:

    • /signup: Allows new users to create an account. Returns a JWT token upon successful registration.
    • /login: Authenticates existing users and returns a JWT token.
    • /todos/ (POST): Create a new todo item. Requires authentication.
    • /todos/ (GET): List all todo items for the authenticated user. Supports pagination with skip and limit.
    • /todos/{todo_id} (PUT): Update a specific todo item. Requires that the todo belongs to the authenticated user.
    • /todos/{todo_id} (DELETE): Delete a specific todo item. Requires that the todo belongs to the authenticated user.

  6. Running the Application:

    • Use Uvicorn to run the FastAPI application. Replace your_file_name with the actual filename of your script.
uvicorn your_file_name:app --reload

The --reload flag enables auto-reloading on code changes, which is useful during development.

Testing the API

You can interact with the API using tools like cURL, Postman, or directly via the interactive API docs provided by FastAPI. Once the server is running, navigate to http://127.0.0.1:8000/docs to access the Swagger UI where you can test all the endpoints.

Security Considerations

  • Secret Key: Ensure that the SECRET_KEY is kept confidential and is sufficiently random. In production, consider loading it from environment variables or a secure vault.
  • HTTPS: Always run your API over HTTPS to ensure that tokens and sensitive data are transmitted securely.
  • Token Expiry & Refresh: The example uses a simple access token. For enhanced security, implement refresh tokens and token revocation mechanisms as needed.
  • Input Validation: While Pydantic models handle basic validation, consider adding more robust validation logic as your application scales.

This example provides a solid foundation for a secured Todo API using FastAPI. You can further enhance it by adding features like user roles, more detailed error handling, and integrating with frontend applications.

Comments

Leave a Reply

Your email address will not be published. Required fields are marked *

More posts