Basic Task - Todo Ex - python + FastAPI + postgres DB

 We will start creating a Todo application using FastAPI and backend with Postgres DB. I am planning to use Python programming language.

for Postgres installation - best option is installing via docker. check the other docs for initial installation steps.

DB commands:

for login: psql -U admin -d windows_fun -W;

for DB creation : CREATE DATABASE todo_db;

connecting DB: \c todo_db;

Table creation: 

    CREATE TABLE todos (

    id SERIAL PRIMARY KEY,
    title VARCHAR(255) NOT NULL,
    completed BOOLEAN DEFAULT FALSE,
    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
)

Table validation: \dt 

Table Field validation: \d todos

SELECT * FROM todos;


============= model.py===================

from from database import get_cursor

def create_todo(title: str):
    cur = get_cursor()

    cur.execute(
    """INSERT INTO todos(title)
    values (%s) RETURNING id, title, completed, created_at
    """,
    (title,), )
   
    todo = cur.fetchone()
    cur.close()
    return todo

def get_all_todo():
    cur = get_cursor
    cur.execute(
        """SELECT id, title, completd, created_at from todos ORDER BY ID"""
    )

    todo = cur.fetchall()
    cur.close()
    return todo


def get_todo(todo_id: int):
    cur = get_cursor
    cur.execute(
        """SELECT id, title, completd, created_at from todos where id %s""", (todo,),
    )

    todo = cur.fetchone()
    cur.close()
    return todo


def updated_tod(todo_id: int,completed: bool):
    cur = get_cursor
    cur.execute(
        """updated todos SET completed=%s, where id=%s RETURNING id, title, completd, created_at""", (completed, todo,),
    )

    todo = cur.fetchone()
    cur.close()
    return todo


============ database.py================

# CREATE TABLE todos (
#        id SERIAL PRIMARY KEY,
#     title VARCHAR(255) NOT NULL,
#     completed BOOLEAN DEFAULT FALSE,
#     created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
# )

# pip install psycopg2-binary
import psycopg2
conn = psycopg2.connect(
    host= "localhost",
    database="todos",
    user = "admin",
    password= "qwerty123"
)

conn.autocommit =True # any query will be autocommitted
def get_cursor():
    return conn.cursor()


============ schema.py=================

from pydnatic import BaseModel

class TodoCreate(BaseModel):
    title: str


class TodoUpdate(BaseModel):
    completed: bool


================main.py===================

from schema import TodoCreate, TodoUpdate
from fastapi import FastAPI, HTTPException
import models

app= FastAPI()

@app.get("/")
def todo():
    return {"message": "todo app running"}

@app.post("/todos")
def create_todo(todoCreate: TodoCreate):
    new_todo: list =models.create_todo(todo.title)
    return{"id": new_todo[0], "title": new_todo[1],
    "completed": new_todo[2], "created_at": new_todo[3]}

@app.get("/todos")
def get_all_todo():
    todos  = models.get_all_todo()
    result = []
    for todo in todos:
        result.append({"id": todo[0], "title": todo[1],
    "completed": todo[2], "created_at": todo[3]    }    )
    return {"message": "todo app running"}

===============================





Comments