Forem: Kapil Bhandari

Your First Journey into RabbitMQ with Python

Kapil Bhandari — Sat, 16 Dec 2023 17:38:32 +0000

Hello World Program

Here we are using pika python client. You can pick anyone as per your application.

Find the list of clients here

Prerequisites

RabbitMQ is installed and is running on localhost. You can simply do this by running the docker image :

   docker run -it --rm --name rabbitmq -p 5672:5672 -p 15672:15672 rabbitmq:3.12-management

Pika Client

pip install pika --upgrade

The Program:

Producer Program - sender.py

import pika

connection = pika.BlockingConnection(
    pika.ConnectionParameters(host='localhost'))
channel = connection.channel()

# INFO: this connects to a broker on the local machine(localhost)

# Creating hello queue

channel.queue_declare(queue="hello")

# INFO: message needs to go through the exchange
# exhange specify exactly to which queue the message should go
# Thus, The queue name needs to be specified in the routing_key parameter:

channel.basic_publish(
    exchange='',
    routing_key='hello',
    body='Hello World')

print("[x] Sent 'Hello World!")


# Before exiting the program we need to make sure the network buffers were flushed and our message was delivered to RabbitMQ. We can do it by gently closing the connection.

connection.close()

Consumer Program - receiver.py

import pika, sys, os


def main():
    connection = pika.BlockingConnection(pika.ConnectionParameters(host='localhost'))
    channel = connection.channel()

    channel.queue_declare(queue='hello')

    # Receiving a message works by subscribing a callback function
    #to a queue  Whenever we receive a message, this callback function is called by the Pika library.

    def callback(ch, method, properties, body):
        print(f" [x] Received {body}")


    channel.basic_consume(queue='hello', on_message_callback=callback, auto_ack=True)
    #finally, we enter a never-ending loop that waits for data and runs callbacks whenever necessary,
    # and catch KeyboardInterrupt during program shutdown.

    print("[*] Waiting for message. To exit  press Ctrl+C")
    channel.start_consuming()

if __name__ == '__main__':
    try:
        main()
    except KeyboardInterrupt:
        print('Interrupted')
        try:
            sys.exit(0)
        except SystemExit:
            os._exit(0)

Running

Run Consumer

python receiver.py

Run Producer

python sender.py

Output

By, Consumer

# => [*] Waiting for messages. To exit press CTRL+C
# => [x] Received 'Hello World!'

Reference

Unravel the secrets of RabbitMQ with our reference to the original RabbitMQ tutorial here.

Conclusion

With this guide, you've taken the first step into a world of endless possibilities.
Stay tuned for more advanced programs

FastAPI & SQLAlchemy: MySQL Setup Guide

Kapil Bhandari — Fri, 15 Dec 2023 09:51:32 +0000

Introduction

In the fast-evolving world of web development, choosing the right tools is crucial for building robust and efficient applications. This article will guide you through the process of setting up FastAPI, a modern, fast, web framework for building APIs with Python, along with SQLAlchemy, a powerful SQL toolkit, and MySQL, a popular open-source relational database management system. By the end of this tutorial, you'll have a solid foundation for developing high-performance web applications.

Guide Overview:

This guide outlines the professional setup of FastAPI in conjunction with SQLAlchemy, utilizing MySQL as the chosen database. By following these steps, you will establish a robust foundation for developing sophisticated and high-performance web applications.

Installation Steps:

To initiate the setup, execute the following commands to install the necessary dependencies:

1. pip install fastapi "uvicorn[standard]"
2. pip install python-dotenv
3. pip install sqlalchemy
4. pip install pymysql

This will install Fastapi and uvicorn to work as the server.

Requirements:

MySql Database

Folder setup:

/app/
    - __init__.py
    - main.py
    - database.py
    - models.py
    - schema.py
    - crud.py
/.env
/requirements.txt

First Let’s create a database engine : app/database.py

app/database.py

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

load_dotenv()
import os
DB_URL = DB_URL = os.getenv("DB_URL")
engine = create_engine(DB_URL,echo=True)
SessionLocal = sessionmaker(autocommit=False,autoflush=False, bind=engine)

Base = declarative_base()

This is the part where we are connecting to MYSQL database

Create database model:

app/models.py

from sqlalchemy import Boolean, Column, ForeignKey, Integer, String
from sqlalchemy.orm import relationship

from .database import Base

class User(Base):
    __tablename__ = "users"
    id = Column(Integer,primary_key=True,index=True)
    name = Column(String(255),index=True)
    email = Column(String(255), unique=True, index=True)
    todos = relationship("Todo",back_populates="owner")
    is_active = Column(Boolean,default=False)

class Todo(Base):
    __tablename__ = "todos"
    id = Column(Integer, primary_key=True, index=True)
    title = Column(String(255), index=True)
    description = Column(String(255), index=True)
    owner_id = Column(Integer, ForeignKey("users.id"))

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

Create **Pydantic models/schemas:**

app/schemas.py

from pydantic import BaseModel

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

class TodoCreate(TodoBase):
    pass

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

    class Config:
        orm_mode = True

class UserBase(BaseModel):
    email: str
    name: str

class UserCreate(UserBase):
    pass 

class User(UserBase):
    id : int
    is_active : bool
    todos : list[Todo] = []

    class Config:
        orm_model = True

Crud Utils:

app/crud.py

from sqlalchemy.orm import Session

from . import models,schemas

def get_user(db: Session, user_id: int):
    return db.query(models.User).filter(models.User.id == user_id).first()

def get_user_by_email(db: Session, email: str):
    return db.query(models.User).filter(models.User.email == email).first()

def get_users(db: Session, skip:int=0, limit:int=100):
    # return db.query(models.User).offset(skip).limit(limit).all()
    return db.query(models.User).offset(skip).limit(limit).all()

def create_user(db: Session, user:schemas.UserCreate):
    db_user = models.User(email=user.email,
                          name=user.name)
    db.add(db_user)
    db.commit()
    db.refresh(db_user)
    return db_user

def get_todos(db: Session, skip:int=0, limit: int=100):
    return db.query(models.Todo).offset(skip).limit(limit).all()

def create_user_todo(db:Session, todo:schemas.TodoCreate, user_id : int):
    db_todo = models.Todo(**todo.model_dump(),owner_id=user_id )
    db.add(db_todo)
    db.commit()
    db.refresh(db_todo)
    return db_todo

# NOTE :
# - add that instance object to your database session.
# - commit the changes to the database (so that they are saved).
# - refresh your instance (so that it contains any new data from the database, like the generated ID).

FastAPI Entry point

app/main.py

from fastapi import FastAPI, Depends, HTTPException
from sqlalchemy.orm import Session

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()

@app.post("/users/",response_model=schemas.User)
def post_user(user:schemas.UserCreate, db:Session=Depends(get_db)):
    db_user = crud.get_user_by_email(db, email=user.email)
    if db_user:
        raise HTTPException(status_code=400, detail="Email already registered")
    return crud.create_user(db=db,user=user)

@app.get("/users/", response_model=list[schemas.User])
def get_users(skip:int=0, limit:int=0, db:Session=Depends(get_db)):
    users = crud.get_users(db,skip=skip,limit=limit)
    return users

@app.get("/users/{user_id}/",response_model=schemas.User)
def get_user(user_id:int, db:Session=Depends(get_db)):
    db_user = crud.get_user(db,user_id =user_id )
    if db_user is None:
        raise HTTPException(status_code=404, detail="User not found")
    return db_user

@app.post("/users/{user_id}/todos/",response_model=schemas.Todo)
def post_todo_for_user(user_id:int, todo:schemas.TodoCreate, db:Session=Depends(get_db)):
    return crud.create_user_todo(db=db,user_id=user_id, todo=todo)

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

Environment Variables:

.env

DB_URL = "mysql+pymysql://{db_username}:{db_password}@localhost:3306/{db_name}"

Result:

uvicorn app.main:app --reload

Conclusion

Congratulations! You've successfully set up a FastAPI application integrated with SQLAlchemy and MySQL.

Thank you 😊

Forem: Kapil Bhandari

Your First Journey into RabbitMQ with Python

Hello World Program

Prerequisites

The Program:

Running

Output

Reference

Conclusion

FastAPI & SQLAlchemy: MySQL Setup Guide

Introduction

Guide Overview:

Installation Steps:

Requirements:

Folder setup:

Create database model:

Create Pydantic models/schemas:

Crud Utils:

FastAPI Entry point

Environment Variables:

Result:

Conclusion

Create **Pydantic models/schemas:**