# Create project folder
# Set-up virtual environment: python3 -m venv venv
# Set venv python interpreter: ./venv/bin/python
# Active venv: source venv/bin/activate
# Install FastAPI: pip install "fastapi[standard]"
# Show all python library: pip freeze
#Get data from body:
from fastapi.params import Body
def create_post(payload: dict = Body(...)):
# Try exception: Internal server error
# Best practic params:
def get_post(id: int):
# Change main.py file directory:
uvicorn app.main:app --reload
# Database:
table: products
id : serial
name: character varaying
price: integer
created_at: time stamp with time zone (default: NOW())
is_sale: boolean
# Key name change:
SELECT id As product_id, is_sale As on_sale FROM products
SELECT * FROM products WHERE id = 10
SELECT * FROM products WHERE price > 50
SELECT * FROM products WHERE inventory > 0 AND price > 20
SELECT * FROM products WHERE id IN (1,2,3)
SELECT * FROM products WHERE name LIKE 'tv%'
SELECT * FROM products WHERE name NOT LIKE 'tv%'
# Order:
SELECT * FROM products ORDER BY price ASC/DESC
SELECT * FROM products WHERE price > 20 ORDER BY created_at
# Limit:
SELECT * FROM products LIMIT 10
SELECT * FROM products WHERE name LIKE 'tv%' LIMIT 5
# Skip: pagination
SELECT * FROM products ORDER BY id LIMIT 10 OFFSET 2
# Insert:
INSERT INTO PRODUCTS (name, price, inventory) VALUES ('tortilla', 4, 100)
INSERT INTO PRODUCTS (name, price, inventory) VALUES ('tortilla', 4, 100), ('Car', 1000, 10)
# Returning:
INSERT INTO PRODUCTS (name, price, inventory) VALUES ('tortilla', 4, 100) RETURNING *
# Delete:
DELETE FROM products WHERE id = 10
DELETE FROM products WHERE inventory > 10
# Update:
UPDATE products SET name = 'flower torilla' , price = 33 WHERE id = 24
#DB Connection:
import psycopg2
from psycopg2.extras import RealDictCursor
import time
# connect db
while True:
try:
conn = psycopg2.connect(
host = 'localhost',
database = 'socialpost',
user = 'postgres',
password = '1234',
cursor_factory= RealDictCursor)
cur = conn.cursor()
print('DB Connected')
break
except Exception as error:
print(f'DB connection error: {error}')
time.sleep(2)
# CURD operation
class Post(BaseModel):
title: Optional[str] = None
content: Optional[str] = None
published: bool = True
@app.get("/")
async def root():
return {"message": "Hello World"}
@app.get("/posts")
async def get_posts():
cursor.execute("""SELECT * FROM posts """)
posts = cursor.fetchall()
print(f'posts: {posts}')
if posts == []:
raise HTTPException(status_code= status.HTTP_404_NOT_FOUND, detail='Empty posts')
return {"message": "Post created successfully", "data" : posts}
@app.post("/posts", status_code = status.HTTP_201_CREATED)
async def create_post(post: Post):
try:
# Works but not safe from hackers
# cursor.execute(f"INSERT INTO posts (title, content, published) VALUES ({post.title}, {post.content},
# {post.published})")
# Working and safe from hackers
cursor.execute("""INSERT INTO posts (title, content, published) VALUES (%s, %s, %s) RETURNING * """,
vars= (post.title, post.content, post.published))
conn.commit()
new_post = cursor.fetchone()
return {"message": "Post created successfully", "data" : new_post}
except Exception as error:
return HTTPException(status_code= status.HTTP_500_INTERNAL_SERVER_ERROR, detail=str(error))
@app.get("/posts/{id}")
async def single_post(id: int):
cursor.execute("""SELECT * FROM posts WHERE id = %s """, (str(id)))
post = cursor.fetchone()
print(f"post: {post}")
if not post:
raise HTTPException(status_code= status.HTTP_404_NOT_FOUND, detail='Post not found')
return {"message": "Post fetched successfully", "data" : post}
@app.delete("/posts/{id}")
async def delete_post(id: int):
cursor.execute("""DELETE FROM posts WHERE id = %s RETURNING * """, (str(id)))
post = cursor.fetchone()
print(f"post: {post}")
if not post:
raise HTTPException(status_code= status.HTTP_404_NOT_FOUND, detail=f'Post not found with {id}')
conn.commit()
return {"message": "Post fetched successfully"}
@app.put("/posts/{id}")
async def update_post(id: int, post: Post):
cursor.execute("""UPDATE posts SET title = %s, content = %s, published = %s WHERE id = %s RETURNING * """,
(post.title, post.content, post.published, str(id)))
updatedPost = cursor.fetchone()
if not updatedPost:
raise HTTPException(status_code= status.HTTP_404_NOT_FOUND, detail=f'Post not found with {id}')
conn.commit()
return {"message": "Post updated successfully", "data" : updatedPost}
# post table to get user info
select * from posts LEFT JOIN users ON posts.owner_id = users.id
#******************** database migration ********************
>pip install alembic
>alembic init alembic
alembic/env.py
from app.models import Base
from app.config import settings
# this is the Alembic Config object, which provides
# access to the values within the .ini file in use.
config = context.config
config.set_main_option(
"sqlalchemy.url", f"postgresql+psycopg2://{settings.database_username}:{settings.database_password}@{settings.database_hostname}:{settings.database_port}/{settings.database_name}")
# Interpret the config file for Python logging.
# This line sets up loggers basically.
if config.config_file_name is not None:
fileConfig(config.config_file_name)
# add your model's MetaData object here
# for 'autogenerate' support
# from myapp import mymodel
# target_metadata = mymodel.Base.metadata
target_metadata = Base.metadata
alembic.ini
sqlalchemy.url =
#create table in db
> alembic revision -m "create posts table"
> alembic upgrade c285645d9c3c
#Add new column
> alembic revision -m 'add content column to posts table'
> alembic upgrade 257574240a34
# delete content column
> alembic downgrade c285645d9c3c
# add new table
> alembic revision -m 'add user table'
> alembic upgrade 82c7bc24978d
# add foreign key
> alembic revision -m 'add foreign-key to post table'
"""add foreign-key to posts table
Revision ID: a2d41404e63b
Revises: 82c7bc24978d
Create Date: 2024-10-14 21:03:03.930077
"""
from typing import Sequence, Union
from alembic import op
import sqlalchemy as sa
# revision identifiers, used by Alembic.
revision: str = 'a2d41404e63b'
down_revision: Union[str, None] = '82c7bc24978d'
branch_labels: Union[str, Sequence[str], None] = None
depends_on: Union[str, Sequence[str], None] = None
def upgrade() -> None:
op.add_column("posts", sa.Column('owner_id', sa.Integer(), nullable=False))
op.create_foreign_key('post_users_fk', source_table='posts', referent_table='users',
local_cols='owner_id', remote_cols='id',ondelete='CASCADE')
pass
def downgrade() -> None:
op.drop_constraint('post_users_fk', table_name='posts')
op.drop_column('posts', 'owner_id')
pass
> alembic heads
> alembic upgrade head
# another add foreign key
> alembic revision -m 'add last few columns to posts table'
"""add last few columns to posts table
Revision ID: 9f392a94f59b
Revises: a2d41404e63b
Create Date: 2024-10-15 21:06:54.362313
"""
from typing import Sequence, Union
from alembic import op
import sqlalchemy as sa
# revision identifiers, used by Alembic.
revision: str = '9f392a94f59b'
down_revision: Union[str, None] = 'a2d41404e63b'
branch_labels: Union[str, Sequence[str], None] = None
depends_on: Union[str, Sequence[str], None] = None
def upgrade() -> None:
op.add_column("posts", sa.Column('published', sa.Boolean(), nullable=False, server_default="TRUE"))
op.add_column("posts", sa.Column('created_at', sa.TIMESTAMP(timezone=True), nullable=False, server_default=sa.text('NOW()')))
pass
def downgrade() -> None:
op.drop_column('posts', 'published')
op.drop_column('posts', 'created_at')
pass
> alembic upgrade +1************************* important ***************************
# auto alembic generate when already create models
> alembic revision --autogenerate -m 'auto-vote'
> alembic upgrade head
************************* important ***************************
************************* CROS ***************************
https://fastapi.tiangolo.com/tutorial/cors/?h=#use-corsmiddleware
#create requirements file
> pip freeze > requirements.txt
# install all dependency from requirements.txt
> pip install -r requirements.txt
# if db update on live server. i used heroku
then
command: heroku run "alembic upgrade head"
# redis install in own machine | brew install redis | then | pip install redis |
then check | redis-server | or | redis-cli
Comments
Post a Comment