Run Necessary Dependencies
python -m venv venv
.\venv\Scripts\activate
C:\Users\rakes\PycharmProjects\fastApiProject>python -m venv venv
C:\Users\rakes\PycharmProjects\fastApiProject>.\venv\Scripts\activate
pip install fastapi
pip install sqlalchemy
pip install psycopg2
pip install psycopg2-binary
pip install pydantic
pip install asyncpg
Setting Up the Environment:
Import Libraries:
from fastapi import FastAPI, HTTPException, Depends
from sqlalchemy import create_engine, Column, Integer, String, Boolean
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker, Session
from sqlalchemy.future import select
from pydantic import BaseModel
These libraries are used for creating the FastAPI application, defining the database models, and handling database sessions.
Database Connection:
DATABASE_URL = "postgresql://postgres:root@localhost:5433/fastapi_database"
engine = create_engine(DATABASE_URL)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
Base = declarative_base()
Here, you're connecting to a PostgreSQL database using SQLAlchemy. The DATABASE_URL contains the connection string for your database. SessionLocal is a session maker that you will use to create sessions to interact with the database.
Define the Task Model:
class Task(Base):
__tablename__ = "tasks"
id = Column(Integer, primary_key=True, index=True)
title = Column(String, nullable=False)
done = Column(Boolean, default=False)
This SQLAlchemy model defines the structure of the tasks table in your database. It has three columns:
id: An integer that serves as the primary key.
title: A string that cannot be null.
done: A boolean that indicates whether the task is completed, with a default value of False.
Create the Database Tables:
Base.metadata.create_all(bind=engine)
This line ensures that the tasks table is created in your database if it doesn't already exist.
Define Pydantic Models for Validation:
These models are used to validate the data you send and receive through the API.
class TaskCreate(BaseModel):
title: str
done: bool = False
class TaskUpdate(BaseModel):
title: str
done: bool
class TaskResponse(BaseModel):
id: int
title: str
done: bool
class Config:
orm_mode = True
TaskCreate: Used when creating a new task.
TaskUpdate: Used when updating an existing task.
TaskResponse: Used when returning data to the client, with orm_mode set to True to allow ORM objects to be returned as responses.
Initialize FastAPI:
app = FastAPI()
This initializes your FastAPI application.
Database Session Dependency:
def get_db():
db = SessionLocal()
try:
yield db
finally:
db.close()
This function provides a database session to each request and ensures that the session is closed after the request is processed.
CRUD Operations:
Create a Task:
@app.post("/tasks/", response_model=TaskResponse)
async def create_task(task: TaskCreate, db: Session = Depends(get_db)):
db_task = Task(title=task.title, done=task.done)
db.add(db_task)
db.commit()
db.refresh(db_task)
return db_task
This endpoint allows you to create a new task. It accepts a JSON body with title and done fields, creates a new Task object, adds it to the database, commits the transaction, and returns the created task.
Example using Postman:
Method: POST
URL: http://localhost:8000/tasks/
Body:
json
Copy code
{
"title": "New Task",
"done": false
}
Read All Tasks:
@app.get("/tasks/", response_model=list[TaskResponse])
async def read_tasks(skip: int = 0, limit: int = 10, db: Session = Depends(get_db)):
tasks = db.execute(select(Task).offset(skip).limit(limit)).scalars().all()
return tasks
This endpoint retrieves a list of tasks, with optional skip and limit query parameters for pagination.
Example using Postman:
Method: GET
URL: http://localhost:8000/tasks/
Read a Task by ID:
@app.get("/tasks/{task_id}", response_model=TaskResponse)
async def read_task(task_id: int, db: Session = Depends(get_db)):
task = db.get(Task, task_id)
if task is None:
raise HTTPException(status_code=404, detail="Task not found")
return task
This endpoint retrieves a task by its ID. If the task is not found, it raises a 404 error.
Example using Postman:
Method: GET
URL: http://localhost:8000/tasks/1
Update a Task:
@app.put("/tasks/{task_id}", response_model=TaskResponse)
async def update_task(task_id: int, task: TaskUpdate, db: Session = Depends(get_db)):
db_task = db.get(Task, task_id)
if db_task is None:
raise HTTPException(status_code=404, detail="Task not found")
db_task.title = task.title
db_task.done = task.done
db.commit()
db.refresh(db_task)
return db_task
This endpoint updates an existing task by its ID. It takes a JSON body with title and done fields.
Example using Postman:
Method: PUT
URL: http://localhost:8000/tasks/1
Body:
{
"title": "Updated Task",
"done": true
}
Delete a Task:
@app.delete("/tasks/{task_id}")
async def delete_task(task_id: int, db: Session = Depends(get_db)):
db_task = db.get(Task, task_id)
if db_task is None:
raise HTTPException(status_code=404, detail="Task not found")
db.delete(db_task)
db.commit()
return {"message": "Task deleted"}
This endpoint deletes a task by its ID. If the task is not found, it raises a 404 error.
Example using Postman:
Method: DELETE
URL: http://localhost:8000/tasks/1
Running the FastAPI Application:
Save the code in a file, for example, main.py.
Run the FastAPI application using the following command:
uvicorn main:app --reload
The application will start, and you can interact with it using the provided URLs.
Interacting with the API using Postman:
You can now use Postman to test the various endpoints:
POST to /tasks/ to create tasks.
GET to /tasks/ to retrieve all tasks.
GET to /tasks/{task_id} to retrieve a task by ID.
PUT to /tasks/{task_id} to update a task.
DELETE to /tasks/{task_id} to delete a task.
By following these steps, you can easily interact with your FastAPI application and manage tasks in your PostgreSQL database.
FULL CODE
from fastapi import FastAPI, HTTPException, Depends
from sqlalchemy import create_engine, Column, Integer, String, Boolean
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker, Session
from sqlalchemy.future import select
from pydantic import BaseModel
DATABASE_URL = "postgresql://postgres:root@localhost:5433/fastapi_database"
# Create a database engine
engine = create_engine(DATABASE_URL)
# Create a SessionLocal class to create session instances
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
# Base class for SQLAlchemy models
Base = declarative_base()
# Define the Task model
class Task(Base):
__tablename__ = "tasks"
id = Column(Integer, primary_key=True, index=True)
title = Column(String, nullable=False)
done = Column(Boolean, default=False)
# Create the database tables
Base.metadata.create_all(bind=engine)
# Pydantic models for creating and updating tasks
class TaskCreate(BaseModel):
title: str
done: bool = False
class TaskUpdate(BaseModel):
title: str
done: bool
class TaskResponse(BaseModel):
id: int
title: str
done: bool
class Config:
orm_mode = True
# Initialize FastAPI
app = FastAPI()
# Dependency to get the database session
def get_db():
db = SessionLocal()
try:
yield db
finally:
db.close()
@app.get("/")
async def root():
return {"message": "Hello World"}
@app.get("/hello/{name}")
async def say_hello(name: str):
return {"message": f"Hello {name}"}
# CRUD Operations
# Create a new task
@app.post("/tasks/", response_model=TaskResponse)
async def create_task(task: TaskCreate, db: Session = Depends(get_db)):
db_task = Task(title=task.title, done=task.done)
db.add(db_task)
db.commit()
db.refresh(db_task)
return db_task
# Read all tasks
@app.get("/tasks/", response_model=list[TaskResponse])
async def read_tasks(skip: int = 0, limit: int = 10, db: Session = Depends(get_db)):
tasks = db.execute(select(Task).offset(skip).limit(limit)).scalars().all()
return tasks
# Read a task by ID
@app.get("/tasks/{task_id}", response_model=TaskResponse)
async def read_task(task_id: int, db: Session = Depends(get_db)):
task = db.get(Task, task_id)
if task is None:
raise HTTPException(status_code=404, detail="Task not found")
return task
# Update a task
@app.put("/tasks/{task_id}", response_model=TaskResponse)
async def update_task(task_id: int, task: TaskUpdate, db: Session = Depends(get_db)):
db_task = db.get(Task, task_id)
if db_task is None:
raise HTTPException(status_code=404, detail="Task not found")
db_task.title = task.title
db_task.done = task.done
db.commit()
db.refresh(db_task)
return db_task
# Delete a task
@app.delete("/tasks/{task_id}")
async def delete_task(task_id: int, db: Session = Depends(get_db)):
db_task = db.get(Task, task_id)
if db_task is None:
raise HTTPException(status_code=404, detail="Task not found")
db.delete(db_task)
db.commit()
return {"message": "Task deleted"}
Top comments (0)