How to Build a Customer Relationship Management System with FastAPI and SendGrid

Time to read: 9 minutes

August 16, 2024

Written by

Samuel Komfi

Contributor

Reviewed by

Diane Phan

Twilion

Amanda Lange

Twilion

Customer Relationship Management (CRM) systems play a pivotal role in helping businesses manage and nurture relationships with their customers. This article will explore the process of building a CRM application using FastAPI, SendGrid for email services, and marketing campaigns to enhance customer engagement and service.

Prerequisites

Before beginning the tutorial, you will need:

A free Sendgrid account.
The latest version of Python 3 installed to your operating system. Follow the instructions for your OS. This tutorial assumes you are developing on a Linux machine.
Your IDE of choice.
A database to store and manage data efficiently. This tutorial will walk you through installation of the PostgreSQL database. However, if you prefer another type of database, install it before beginning this tutorial.

Authentication has been removed from the following article to slim it down somewhat. Other models and schemas are also not shown. However, they can be found in the complete source code hosted on GitHub.

Set Up Your Project

For this tutorial, you will utilize FastAPI. FastAPI is a modern, fast web framework for building APIs with Python 3.7+, and will serve as the backbone of your CRM application. Its simplicity, speed, and automatic OpenAPI and JSON Schema generation make it an excellent choice.

Create a project folder and navigate to it:

mkdir fastcrm
cd fastcrm

Create a virtual environment and immediately activate it:

python -m venv venv
source venv/bin/activate

Now you will install some tools necessary for your application. Pydantic will be used to define models for the application. These models will serve as a bridge between the database and the FastAPI application, providing data validation and serialization.

SQLAlchemy, an SQL toolkit and Object-Relational Mapping (ORM) library, will be used to interact with the database. Alembic, a database migration tool, will help you manage database schema changes over time.

Install the necessary packages in the command line:

pip install 'fastapi[all]' sqlalchemy psycopg2-binary sendgrid 'pydantic[email,dotenv]'

Please note the [all] in fastapi[all] will also install uvicorn and pydantic for you. Pydantic has two optional dependencies for email validation and dotenv. You can install them directly with pip install email-validator python-dotenv or use the command above to install them along with pydantic by enclosing them in brackets. Use whichever works for your system.

Create the folder structure needed for the app. Creating a folder structure is crucial for maintaining an organized, scalable, and maintainable codebase. It separates the different parts of the application, enhancing readability and manageability. You will start by creating the main app/ folder to hold all the code necessary for the project.

mkdir app
cd app
mkdir api crud db models schemas services 
touch main.py .env settings.py __init__.py

The project structure will follow a modular approach, with separate folders for models, schemas, CRUD operations, services, and API endpoints. It will be based on the following structure:

Models ( app/models/) - Define SQLAlchemy models for the CRM entities - Leads, Customers, Users, Emails, Campaigns, and Deals.

Schemas ( app/schemas/) - Pydantic models for data validation and serialization will be created for each entity.

CRUD Operations ( app/crud/) - These files contain CRUD (Create, Read, Update, Delete) operations for each entity. They serve as the intermediary between the database and the API.

Services ( app/services/) - The EmailService class, responsible for sending emails through SendGrid and tracking email opens, is placed in this folder.

API Endpoints ( app/api/) - FastAPI routes for each entity, including endpoints for basic CRUD operations and email-related functionalities.

When it is complete, the model for the project database will look like the following:

Create the PostgreSQL Database and user

If you are choosing PostgreSQL for your database, install it now using the instructions for your appropriate operating system. Otherwise, you can use the database of your choice and move onto the next segment of the tutorial.

Install PostgreSQL on Windows

Step 1: Download PostgreSQL Installer

- Go to the official PostgreSQL download page.

- Click on the "Download the installer" link.

- Choose the version you want and download the installer for Windows.

Step 2: Run the Installer

1. Run the downloaded installer file.

2. Follow the setup wizard. Click "Next" to proceed with the default options or customize the installation directory if needed.

Step 3: Choose Installation Directory

Choose the directory where you want PostgreSQL to be installed. Click "Next".

Step 4: Select Components

Select the components you want to install. Typically, you'll want to install PostgreSQL Server, pgAdmin 4, and Command Line Tools. Click "Next".

Step 5: Set Password

Set a password for the PostgreSQL superuser (default user is postgres). Make sure to remember this password. Click "Next".

Step 6: Set Port Number

Set the port number for PostgreSQL. The default port is 5432. Click "Next".

Step 7: Choose Locale

Choose the locale settings. By default, it is set to your system's locale. Click "Next".

Step 8: Start Installation

Click "Next" and then "Finish" to start the installation process. Wait for the installation to complete.

Step 9: Verify Installation

Open the pgAdmin 4 tool from the Start menu.

Connect to the PostgreSQL server using the password you set during installation.

Install PostgreSQL on macOS

Step 1: Install Homebrew

1. Open Terminal.

2. Install Homebrew if you haven't already:

/bin/bash -c "$(curl -fsSL https://raw.githubusercontent.com/Homebrew/install/HEAD/install.sh)"

Step 2: Install PostgreSQL

Use Homebrew to install PostgreSQL:

brew install postgresql

Step 3: Start PostgreSQL

Start the PostgreSQL service:

brew services start postgresql

Step 4: Verify Installation

Verify the installation by checking the PostgreSQL version:

psql --version

Step 5: Access PostgreSQL

Access the PostgreSQL prompt:

psql postgres

Install PostgreSQL on Linux

Step 1: Update Package List

Update your package list:

sudo apt update

Step 2: Install PostgreSQL

Install PostgreSQL:

sudo apt install postgresql postgresql-contrib

Step 3: Start PostgreSQL Service

Ensure the PostgreSQL service is running:

sudo systemctl start postgresql

Step 4: Enable PostgreSQL Service

Enable PostgreSQL to start on boot:

sudo systemctl enable postgresql

Step 5: Switch to PostgreSQL User

Switch to the PostgreSQL user:

sudo -i -u postgres

Step 6: Access PostgreSQL Prompt

Access the PostgreSQL prompt:

psql

Step 7: Exit PostgreSQL Prompt

Exit the PostgreSQL prompt:

\q

You will need to create a database for the application together with the user. After installing the database login into your database prompt:

sudo -iu postgres psql

Enter the following command to create a new database. For this project, the database can be named "sendgridapp":

CREATE DATABASE sendgridapp;

Create a new user and grant that user all privileges to the newly created database:

CREATE USER dbuser WITH PASSWORD 'dbpassword';
GRANT ALL PRIVILEGES ON DATABASE sendgridapp TO dbuser;

Build Your Python Application

The main.py file will be your entry point to the application. The settings.py file is used to get environment variables from an .env file and store them in the Settings class.

Create an .env file and add the environment variables needed, which is the Sendgrid API Key and the Database URL. Use the database details above to fill in the dbname, dbuser, dbserver and dbpassword in the URL.

SENDGRID_API_KEY=”YOUR_SENDGRID_API_KEY”
DATABASE_URL=”postgresql://{DBUSER}:{DBPASSWORD}@{DBSERVER}/{DBNAME}”

The "DBUSER" and "DBPASSWORD" correspond to the database login you use for your database. For instance the DBNAME will be the name of the database hosted on the DBSERVER. If you used the setup above, the default value for DBSERVER is localhost.

After creating the file add the following to settings.py:

import os
from typing import Optional
from pydantic_settings import BaseSettings
# Specify where the environment file is
DOTENV = os.path.join(os.path.dirname(__file__), ".env")
class Settings(BaseSettings):
    sendgrid_api_key: Optional[str] = None
    database_url: Optional[str] = None 
    twilio_account_sid: Optional[str] = None
    twilio_auth_token: Optional[str] = None
    class Config:
        env_file = DOTENV  
# Create an instance of Settings to be accessed 
settings = Settings()

Currently, the directory structure of your application should look like the following, with the exception of the alembic/ folder which will be created with the command alembic:

Before creating the database models with SQLAlchemy, add the connection to the database by creating a new file in the /db folder called database.py. This is the contents of the new db/database.py file:

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker, declarative_base
from settings import settings
DATABASE_URI = settings.database_url
engine = create_engine(DATABASE_URI)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
Base = declarative_base()
# Dependency to get the database session
def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

Create the Database Models

Now, create the database models.

cd models
touch lead.py customer.py user.py email.py campaign.py deal.py __init__.py

The models in the lead.py file will look like the following:

from sqlalchemy import Column, DateTime, Integer, String, func
from sqlalchemy.orm import relationship
from db.database import Base
class Lead(Base):
    __tablename__ = "leads"
    id = Column(Integer, primary_key=True, index=True)
    name = Column(String(60), index=True, nullable=False)
    email = Column(String, index=True)
    phone = Column(String)
    status = Column(String)
    created_at = Column(DateTime, default=func.now())
    updated_at = Column(DateTime, default=func.now())
    deal = relationship("Deal", back_populates="lead")
    message = relationship("Email", back_populates="lead")

Model file for Deals( deal.py):

from sqlalchemy import Column, DECIMAL, Integer, String, ForeignKey
from sqlalchemy.orm import relationship
from db.database import Base
class Deal(Base):
    __tablename__ = "deals"
    id = Column(Integer, primary_key=True, index=True)
    name = Column(String, index=True, nullable=False)
    amount = Column(DECIMAL(10, 2))
    stage = Column(String)
    lead_id = Column(Integer, ForeignKey("leads.id"))
    customer_id = Column(Integer, ForeignKey("customers.id"))
    lead = relationship("Lead", back_populates="deal")
    customer = relationship("Customer", back_populates="deal")

The customer.py model file:

from sqlalchemy import Column, DateTime, Integer, String, func
from sqlalchemy.orm import relationship
from db.database import Base
class Customer(Base):
    __tablename__ = "customers"
    id = Column(Integer, primary_key=True, index=True)
    name = Column(String, index=True, nullable=False)
    email = Column(String, index=True)
    phone = Column(String)
    created_at = Column(DateTime, default=func.now())
    updated_at = Column(DateTime, default=func.now())
    deal = relationship("Deal", back_populates="customer")

The campaign model file( campaign.py):

from sqlalchemy import Column, DateTime, Integer, String, func
from sqlalchemy.orm import relationship
from db.database import Base
class Campaign(Base):
    __tablename__ = "campaigns"
    id = Column(Integer, primary_key=True, index=True)
    name = Column(String, index=True, nullable=False)
    html_content = Column(String)
    sendgrid_campaign_id = Column(String)
    start_date = Column(DateTime)
    end_date = Column(DateTime)
    description = Column(String)

The email.py file that will store the emails locally. Each email will belong to a lead and each email will store a SendGrid email_id to link it. Here is the file:

from sqlalchemy import Boolean, Column, DateTime, Integer, String, ForeignKey, func
from sqlalchemy.orm import relationship
from db.database import Base
class Email(Base):
    __tablename__ = "emails"
    id = Column(Integer, primary_key=True, index=True)
    subject = Column(String)
    body = Column(String)
    sender = Column(String, nullable=False)
    recipient = Column(String, nullable=False)
    sendgrid_email_id = Column(String)
    sent_at = Column(DateTime, default=func.now())
    is_opened = Column(Boolean, default=False)
    lead_id = Column(Integer, ForeignKey("leads.id"))
    lead = relationship("Lead", back_populates="message")

And finally the user.py model file for the sales rep or users:

from sqlalchemy import Column, DateTime, ForeignKey, Integer, String, func
from sqlalchemy.orm import relationship
from db.database import Base
class User(Base):
    __tablename__ = "users"
    id = Column(Integer, primary_key=True, index=True)
    username = Column(String, unique=True, index=True, nullable=False)
    full_name = Column(String, index=True)
    email = Column(String, unique=True, index=True, nullable=False)
    password = Column(String, nullable=False)
    created_at = Column(DateTime, default=func.now())
    updated_at = Column(DateTime, default=func.now())

Use the Alembic Tool to Manage Database Migrations

After creating the models it is time to apply the schema to the database. Use alembic to manage the migrations.

pip install alembic

Run the following to command from the /app directory to initialize an alembic managed project and specify the directory (usually also named alembic) for alembic migration files:

alembic init alembic

The command generates a file called alembic.ini in the current folder along with a directory to hold the migration changes using whatever name is given.

Edit the env.py file in the alembic/ folder to include your models and import the sqlalchemy metadata object by adding this code to the top:

# add your model's MetaData object here
# for 'autogenerate' support
# from fastcrm import mymodel
# target_metadata = mymodel.Base.metadata
from models.campaign import Campaign
from models.customer import Customer
from models.deal import Deal
from models.email import Email
from models.lead import Lead
from models.user import User
from db.database import Base
target_metadata = Base.metadata

Also edit the alembic.ini file generated by specifying the sqlalchemy url which is the same as your database URL.

Look for the line sqlalchemy.url = driver://user:pass@localhost/dbname in your code, and replace it with correct and updated information for your database, replacing the placeholders as needed:

sqlalchemy.url = postgresql://[dbuser]:[dbpassword]@[dbserver]/[dbname]

To create tables in the database, generate migrations from your created models and run the following alembic command, which will generate the SQL to commit to the database:

alembic revision --autogenerate -m "Create tables for App"

To complete the process and execute the SQL generated upgrade the migration in the command line:

alembic upgrade head

Create the functions for CRUD operations

Change into the crud/ folder. The crud/ folder is like a controller for your app. It allows you to create functions that will make the necessary calls to the database or service. Create the necessary files:

touch crud_campaign.py crud_customer.py crud_deal.py crud_email.py crud_lead.py crud_user.py

Here is sample of the crud/crud_lead.py file:

from sqlalchemy.orm import Session
from models.lead import Lead
from schemas.lead import LeadCreate
def create_lead(db: Session, name: str, email: str, phone: str, status: str):
    db_lead = Lead(name=name, email=email, phone=phone, status=status)
    db.add(db_lead)
    db.commit()
    db.refresh(db_lead)
    return db_lead
def get_lead(db: Session, lead_id: int):
    return db.query(Lead).filter(Lead.id == lead_id).first()
def get_all_leads(db: Session, skip: int = 0, limit: int = 10):
    return db.query(Lead).offset(skip).limit(limit).all()

Now create the CRUD operations for the users in the crud_users.py file:

from sqlalchemy.orm import Session
from models.user import User
from schemas.user import UserCreate
def create_user(db: Session, username: str, email: str, full_name: str, password: str):
    db_user = User(
        username=username,
        email=email,
        full_name=full_name,
        password=password,
    )
    db.add(db_user)
    db.commit()
    db.refresh(db_user)
    return db_user
def get_user(db: Session, user_id: int):
    return db.query(User).filter(User.id == user_id).first()
def get_user_by_email(db: Session, email: str):
    user = db.query(User).filter(User.email == email).first()
    return user
def get_all_users(db: Session, skip: int = 0, limit: int = 10):
    return db.query(User).offset(skip).limit(limit).all()

Create the CRUD operations for the crud_campaigns.py file :

from datetime import datetime
from sqlalchemy.orm import Session
from models.campaign import Campaign
from schemas.campaign import CampaignCreate
def add_campaign(
    db: Session,
    name: str,
    html_content: str,
    sendgrid_campaign_id: str,
    start_date: datetime,
    end_date: datetime,
    description: str,
):
    db_campaign = Campaign(
        name=name,
        html_content=html_content,
        sendgrid_campaign_id=sendgrid_campaign_id,
        start_date=start_date,
        end_date=end_date,
        description=description,
    )
    db.add(db_campaign)
    db.commit()
    db.refresh(db_campaign)
    return db_campaign
def get_campaign(db: Session, campaign_id: int):
    return db.query(Campaign).filter(Campaign.id == campaign_id).first()
def get_all_campaigns(db: Session, skip: int = 0, limit: int = 10):
    return db.query(Campaign).offset(skip).limit(limit).all()

Create the CRUD operations for the crud_customers.py crud file:

from sqlalchemy.orm import Session
from models.customer import Customer
from schemas.customer import CustomerCreate
def add_customer(db: Session, name: str, email: str, phone: str):
    db_customer = Customer(name=name, email=email, phone=phone)
    db.add(db_customer)
    db.commit()
    db.refresh(db_customer)
    return db_customer
def get_customer(db: Session, customer_id: int):
    return db.query(Customer).filter(Customer.id == customer_id).first()
def get_customer_by_email(db: Session, email: str):
    return db.query(Customer).filter(Customer.email == email).first()
def get_all_customers(db: Session, skip: int = 0, limit: int = 10):
    return db.query(Customer).offset(skip).limit(limit).all()

The code for the crud_deals.py file is shown below:

from sqlalchemy.orm import Session
from models.customer import Customer
from schemas.customer import CustomerCreate
def add_customer(db: Session, name: str, email: str, phone: str):
    db_customer = Customer(name=name, email=email, phone=phone)
    db.add(db_customer)
    db.commit()
    db.refresh(db_customer)
    return db_customer
def get_customer(db: Session, customer_id: int):
    return db.query(Customer).filter(Customer.id == customer_id).first()
def get_customer_by_email(db: Session, email: str):
    return db.query(Customer).filter(Customer.email == email).first()
def get_all_customers(db: Session, skip: int = 0, limit: int = 10):
    return db.query(Customer).offset(skip).limit(limit).all()

And here's the code for crud_emails.py:

from sqlalchemy.orm import Session
from models.email import Email
from schemas.email import EmailCreate
def create_email(
    db: Session, subject: str, body: str, sender: str, recipient: str, lead_id: int
):
    db_email = Email(
        subject=subject,
        body=body,
        sender=sender,
        recipient=recipient,
        lead_id=lead_id,
    )
    db.add(db_email)
    db.commit()
    db.refresh(db_email)
    return db_email
def get_email(db: Session, email_id: int):
    return db.query(Email).filter(Email.id == email_id).first()
def get_all_emails(db: Session, skip: int = 0, limit: int = 10):
    return db.query(Email).offset(skip).limit(limit).all()

Create Pydantic Schemas for data validation

Now you will create some pydantic models which will be in the schemas/ directory. You will need to create new files for all the schemas. Pydantic schemas provide a powerful way to manage and validate data in Python applications. Each schema file will correspond to a model file with the same name for SQLAlchemy.

Here is the campaign.py schema:

from typing import List, Optional
from datetime import datetime
from pydantic import BaseModel, Field
class CampaignBase(BaseModel):
    name: str
    html_content: Optional[str] = None
    start_date: Optional[datetime] = None
    end_date: Optional[datetime] = None
    description: Optional[str] = None
    sendgrid_campaign_id: Optional[str] = None
class CampaignCreate(CampaignBase):
    pass
class Campaign(CampaignBase):
    id: int
    created_at: datetime = Field(default_factory=datetime.now)
    class Config:
        from_attributes = True

The deal.py file schema:

from typing import List, Optional
from datetime import datetime
from pydantic import BaseModel, Field
class DealBase(BaseModel):
    name: str
    amount: Optional[float] = None
    stage: Optional[str] = None
    lead_id: Optional[int] = None
    customer_id: Optional[int] = None
class DealCreate(DealBase):
    pass
class Deal(DealBase):
    id: int
    created_at: datetime = Field(default_factory=datetime.now)
    updated_at: datetime = Field(default_factory=datetime.now)
    class Config:
        from_attributes = True

The email.py schema which will be used by the email_service.py:

from typing import List, Optional
from datetime import datetime
from pydantic import BaseModel, EmailStr
class EmailBase(BaseModel):
    subject: Optional[str] = None
    body: Optional[str] = None
    sender: EmailStr
    recipient: EmailStr
    sendgrid_email_id: Optional[str] = None
    sent_at: Optional[datetime] = None
    is_opened: Optional[bool] = False
    lead_id: Optional[int] = None
class EmailCreate(EmailBase):
    Pass
class Email(EmailBase):
    id: int
    class Config:
        from_attributes = True

The schema for the customer.py will be light as only the name, email and phone are needed:

from typing import List, Optional
from datetime import datetime
from pydantic import BaseModel, EmailStr
class CustomerBase(BaseModel):
    name: str
    email: Optional[EmailStr] = None
    phone: Optional[str] = None
class CustomerCreate(CustomerBase):
    pass
class Customer(CustomerBase):
    id: int
    created_at: Optional[datetime] = None
    updated_at: Optional[datetime] = None
    class Config:
        from_attributes = True

Most of the lead.py fields are optional with the only the name of the lead required:

from typing import List, Optional
from datetime import datetime
from pydantic import BaseModel, EmailStr
class LeadBase(BaseModel):
    name: str
    email: Optional[EmailStr] = None
    phone: Optional[str] = None
    status: Optional[str] = None
class LeadCreate(LeadBase):
    pass
class Lead(LeadBase):
    id: int
    created_at: Optional[datetime] = None
    updated_at: Optional[datetime] = None
    class Config:
        from_attributes = True

And finally add the user.py schema file, none of its fields are optional:

from typing import List, Optional
from datetime import datetime
from pydantic import BaseModel, EmailStr
class UserBase(BaseModel):
    username: str
    email: EmailStr
    full_name: str
class UserCreate(UserBase):
    password: str
class User(UserBase):
    id: int
    created_at: Optional[datetime] = None
    updated_at: Optional[datetime] = None
    class Config:
        from_attributes = True

Send emails with SendGrid

The integration of SendGrid adds a powerful email component to your CRM application. In the EmailService class, you have methods for sending emails, tracking email opens, and retrieving the open status of an email.

The send_email method utilizes SendGrid's Python library to send emails. It leverages the SendGrid API key and stores the email in the database.

Track Status of Open Emails

The track_email_open method updates the is_opened field in the database when an email is opened. This functionality provides insights into customer engagement. The get_email_open_status method retrieves the open status of a specific email. This information can be valuable for follow-up actions.

Create the email_service.py file inside the services/ directory:

cd services
touch email_service.py

Here is the code for email_service.py:

from sendgrid import SendGridAPIClient, TwilioEmailAPIClient
from sendgrid.helpers.mail import Mail
from fastapi import HTTPException
from sqlalchemy.orm import Session
from models.email import Email
from schemas.email import EmailCreate
from settings import settings
class EmailService:
    @staticmethod
    def send_email(api_key: str, email_data: EmailCreate, db: Session):
        sg = SendGridAPIClient(api_key)
        message = Mail(
            from_email=email_data.sender,
            to_emails=email_data.recipient,
            subject=email_data.subject,
            plain_text_content=email_data.body,
            html_content=email_data.body,
        )
        # message.tracking_settings = {"open_tracking": {"enable": True}}
        try:
            response = sg.send(message)
            if response.status_code == 202:
                email_id = response.headers.get("X-Message-Id")
                email_data.sendgrid_email_id = email_id
                email_model = Email(**email_data.dict())
                db.add(email_model)
                db.commit()
                db.refresh(email_model)
                return email_model
            else:
                raise HTTPException(
                    status_code=response.status_code, detail="Failed to send email"
                )
        except Exception as e:
            raise HTTPException(
                status_code=500, detail=f"Error sending email: {str(e)}"
            )
    @staticmethod
    def track_email_open(sendgrid_email_id: str, db: Session):
        sg = SendGridAPIClient(api_key=settings.sendgrid_api_key)
        is_opened = False
        try:
            response = sg.client.campaigns.stats.get(
                query_params={"id": sendgrid_email_id}
            )
            if response.status_code == 200:
                data = response.body
                # Check if the email has been opened, blocks will be empty if email opened
                if data.get("stats"):
                    if not data["stats"].get("blocks"):
                        is_opened = True
                else:
                    print(
                        f"Failed to get email stats. Status Code: {response.status_code}"
                    )
        except Exception as e:
            print(f"An error occured: {e}")
            is_opened = False
        email = (
            db.query(Email).filter(Email.sendgrid_email_id == sendgrid_email_id).first()
        )
        if is_opened:
            email.is_opened = True
            db.commit()
    @staticmethod
    def get_email_open_status(email_id: int, db: Session):
        email = db.query(Email).filter(Email.id == email_id).first()
        if email:
            return {"email_id": email_id, "is_opened": email.is_opened}
        else:
            raise HTTPException(status_code=404, detail="Email not found")

While this CRM application lays the foundation for managing customer relationships, integrating marketing campaigns will further enhance customer engagement and conversion. In your application, you've introduced a Campaign entity and corresponding CRUD operations.

Test the CRM API endpoints

The last folder in the app is the api/ folder which contains the routes of your application. It has routes for each module of your application. Create the endpoint files in the api/ directory:

touch campaign.py customer.py deal.py email.py lead.py user.py __init__.py

For instance the api/lead.py will have 3 endpoints which can be tested:

from typing import List
from fastapi import APIRouter, Depends, HTTPException
from sqlalchemy.orm import Session
from schemas import lead as lead_schemas
from crud import crud_leads
from db.database import get_db
router = APIRouter()
@router.post("/leads/", response_model=lead_schemas.Lead)
def create_lead(lead: lead_schemas.LeadCreate, db: Session = Depends(get_db)):
    return crud_leads.create_lead(
        db=db, name=lead.name, phone=lead.phone, email=lead.email, status=lead.status
    )
@router.get("/leads/{lead_id}", response_model=lead_schemas.Lead)
def read_lead(lead_id: int, db: Session = Depends(get_db)):
    db_lead = crud_leads.get_lead(db=db, lead_id=lead_id)
    if db_lead is None:
        raise HTTPException(status_code=404, detail="Lead not found")
    return db_lead
@router.get("/leads/", response_model=List[lead_schemas.Lead])
def read_leads(skip: int = 0, limit: int = 10, db: Session = Depends(get_db)):
    return crud_leads.get_all_leads(db=db, skip=skip, limit=limit)

The email.py file will retrieve the SENDGRID key set from the .env file:

import os
from typing import List
from fastapi import APIRouter, Depends, HTTPException
from sqlalchemy.orm import Session
from schemas import email as email_schemas
from crud import crud_emails
from db.database import get_db
from services.email_service import EmailService
from settings import settings
router = APIRouter()
api_key = settings.sendgrid_api_key
@router.post("/emails/", response_model=email_schemas.Email)
def send_email(email: email_schemas.EmailCreate, db: Session = Depends(get_db)):
    return EmailService.send_email(api_key=api_key, email_data=email, db=db)
@router.post("/emails/{email_id}/track-open")
def track_email_open(sendgrid_email_id: str, db: Session = Depends(get_db)):
    EmailService.track_email_open(sendgrid_email_id=sendgrid_email_id, db=db)
    return {"message": "Email open tracked successfully"}
@router.get("/emails/{email_id}/open-status", response_model=dict)
def get_email_open_status(email_id: int, db: Session = Depends(get_db)):
    return EmailService.get_email_open_status(email_id=email_id, db=db)

Here is the campaign.py file for campaigns:

from typing import List
from fastapi import APIRouter, Depends
from sqlalchemy.orm import Session
from schemas import campaign as campaign_schemas
from crud import crud_campaigns
from db.database import get_db
router = APIRouter()
@router.post("/campaigns/", response_model=campaign_schemas.Campaign)
def create_campaign(
    campaign: campaign_schemas.CampaignCreate, db: Session = Depends(get_db)
):
    return crud_campaigns.add_campaign(db=db, **campaign.model_dump())
@router.get("/campaigns/{campaign_id}", response_model=campaign_schemas.Campaign)
def read_campaign(campaign_id: int, db: Session = Depends(get_db)):
    db_campaign = crud_campaigns.get_campaign(db=db, campaign_id=campaign_id)
    if db_campaign is None:
        raise HTTPException(status_code=404, detail="Campaign not found")
    return db_campaign
@router.get("/campaigns/", response_model=List[campaign_schemas.Campaign])
def read_campaigns(skip: int = 0, limit: int = 10, db: Session = Depends(get_db)):
    return crud_campaigns.get_all_campaigns(db=db, skip=skip, limit=limit)

The customer.py will handle the API calls for creating and listing customers:

from typing import List
from fastapi import APIRouter, Depends, HTTPException
from sqlalchemy.orm import Session
from schemas import customer as customer_schemas
from crud import crud_customers
from db.database import get_db
router = APIRouter()
@router.post("/customers/", response_model=customer_schemas.Customer)
def create_customer(
    customer: customer_schemas.CustomerCreate, db: Session = Depends(get_db)
):
    db_cust = crud_customers.get_customer_by_email(db, email=customer.email)
    if db_cust:
        raise HTTPException(
            status_code=400, detail="Customer with that email already exists!"
        )
    return crud_customers.add_customer(
        db=db, name=customer.name, email=customer.email, phone=customer.phone
    )
@router.get("/customers/{customer_id}", response_model=customer_schemas.Customer)
def read_customer(customer_id: int, db: Session = Depends(get_db)):
    db_customer = crud_customers.get_customer(db=db, customer_id=customer_id)
    if db_customer is None:
        raise HTTPException(status_code=404, detail="Customer not found")
    return db_customer
@router.get("/customers/", response_model=List[customer_schemas.Customer])
def read_customers(skip: int = 0, limit: int = 10, db: Session = Depends(get_db)):
    return crud_customers.get_all_customers(db=db, skip=skip, limit=limit)

You can now add deal.py to add API functions to handle deal creation and listing:

from typing import List
from fastapi import APIRouter, Depends, HTTPException
from sqlalchemy.orm import Session
from schemas import deal as deal_schemas
from crud import crud_deals
from db.database import get_db
router = APIRouter()
@router.post("/deals/", response_model=deal_schemas.Deal)
def create_deal(deal: deal_schemas.DealCreate, db: Session = Depends(get_db)):
    return crud_deals.create_deal(db=db, **deal.dict())
@router.get("/deals/{deal_id}", response_model=deal_schemas.Deal)
def read_deal(deal_id: int, db: Session = Depends(get_db)):
    db_deal = crud_deals.get_deal(db=db, deal_id=deal_id)
    if db_deal is None:
        raise HTTPException(status_code=404, detail="Deal not found")
    return db_deal
@router.get("/deals/", response_model=List[deal_schemas.Deal])
def read_deals(skip: int = 0, limit: int = 10, db: Session = Depends(get_db)):
    return crud_deals.get_all_deals(db=db, skip=skip, limit=limit)

The remaining user.py file is about handling user related functionality, including checking if a user with certain email exists before proceeding:

from typing import List
from fastapi import APIRouter, Depends, HTTPException
from sqlalchemy.orm import Session
from schemas import deal as deal_schemas
from crud import crud_deals
from db.database import get_db
router = APIRouter()
@router.post("/deals/", response_model=deal_schemas.Deal)
def create_deal(deal: deal_schemas.DealCreate, db: Session = Depends(get_db)):
    return crud_deals.create_deal(db=db, **deal.dict())
@router.get("/deals/{deal_id}", response_model=deal_schemas.Deal)
def read_deal(deal_id: int, db: Session = Depends(get_db)):
    db_deal = crud_deals.get_deal(db=db, deal_id=deal_id)
    if db_deal is None:
        raise HTTPException(status_code=404, detail="Deal not found")
    return db_deal
@router.get("/deals/", response_model=List[deal_schemas.Deal])
def read_deals(skip: int = 0, limit: int = 10, db: Session = Depends(get_db)):
    return crud_deals.get_all_deals(db=db, skip=skip, limit=limit)

The main.py file for the in the app/ folder will be nothing special and will look like most FastAPI applications:

import fastapi
import uvicorn
from db.database import SessionLocal, engine
from api.campaign import router as campaign_router
from api.customer import router as customer_router
from api.deal import router as deal_router
from api.email import router as email_router
from api.lead import router as lead_router
from api.user import router as user_router
app = fastapi.FastAPI()
app.include_router(campaign_router)
app.include_router(customer_router)
app.include_router(deal_router)
app.include_router(lead_router)
app.include_router(email_router)
app.include_router(user_router)
if __name__ == "__main__":
    uvicorn.run("main:app", port=8081, host="127.0.0.1", reload=True)

In the project app/ folder, run the FastAPI application using the command line:

python main.py

You will be greeted with the following confirmation on success:

Navigate to http://127.0.0.1:8081/docs to test the application. The URL points to a Swagger API documentation of the application. It lists all the endpoints for the application which you can test on the same page while it calls curl in the background:

Click on any endpoint you wish to test to reveal a drop down with a button on the right corner that says "Try it out". Click on the button which reveals an "Execute" button together with options and or requirements for your endpoint. The result of the request are displayed below in the response section:

What's next for SendGrid CRM Applications?

SendGrid, a cloud-based email delivery service, enables us to send and track emails seamlessly. It provides a reliable and scalable solution for managing email communication with customers.Building a CRM application shows just how sending emails and building marketing campaigns can help a business thrive with its customer engagement efforts. The modular project structure ensures maintainability, and the use of Pydantic models enhances data validation.

The source code for the complete api can be found here https://github.com/samaras/fastapi-conversation.

You can add generative AI to the application by exploring other Sendgrid APIs like event web hooks to further engage your customers by sending follow up emails on opened leads.

Samuel Komfi is a Freelance Software Developer based in Johannesburg. He has been programming since he was 16 and now shares his thoughts on various blogs. He loves coffee, sunny days and blazingly fast internet to crawl the web for all things not cat videos. He can be reached at skomfi [at] gmail.com.

His profile at X is @skomfi.

Related Resources

Twilio Docs

From APIs to SDKs to sample apps

API reference documentation, SDKs, helper libraries, quickstarts, and tutorials for your language and platform.

Resource Center

The latest ebooks, industry reports, and webinars

Learn from customer engagement experts to improve your own communication.

Ahoy

Twilio's developer community hub

Best practices, code samples, and inspiration to build communications and digital engagement experiences.

How to Build a Customer Relationship Management System with FastAPI and SendGrid

Related Posts

Related Resources

From APIs to SDKs to sample apps

The latest ebooks, industry reports, and webinars

Twilio's developer community hub