Creating an API with python: Part 2: MariaDB Database

In my previous post, Creating an API with python: Part 1: GET Endpoints, I set up an API using FastAPI with five GET endpoints. In this post, I’ll install a MariaDB database and connect the five GET endpoints to the database.

Prerequisites

Creating an API with python: Part 1: GET Endpoints

Step 1: Install MariaDB

I have chosen to use MariaDB for the API database, as it is the native database for CentOS7 and is an open-source fork of the MySQL database.

SSH to your virtual machine/API host.
Run the following to install MariaDB:
```
$ sudo yum install mariadb-server
```
Then run these commands to start and enable MariaDB to run at boot:
```
$ sudo systemctl start mariadb
$ sudo systemctl enable mariadb
```
Next, run the following:
```
$ sudo mysql_secure_installation
```
You will get prompted to:
1. Set a root password
2. Remove anonymous users
3. Disallow root login remotely
4. Remove the test database
5. Reload the privilege tables
It is recommended to answer ‘Y’ to all questions.

Step 2: Set Up Database Users and Tables

Connect to MariaDB from the command line, using the root password you just set up:
```
mysql -u root -p
```
You should now see the MariaDB prompt. Add a user with the command below. Replace PASSWORD with a suitable password, surrounded by single quotes.
```
MariaDB [(none)]> CREATE USER 'apiservice_user'@'localhost' IDENTIFIED BY 'PASSWORD';
```

Create a database called apiservice:

MariaDB [(none)]> CREATE DATABASE IF NOT EXISTS apiservice;

Allow access to the database for the user. Note that the apiservice_user user is intended to be the one the API uses. Therefore it needs access to SELECT, INSERT, UPDATE and DELETE on tables within the apiservice database only:
```
MariaDB [(none)]> GRANT SELECT, INSERT, UPDATE, DELETE ON apiservice.* TO apiservice_user@'localhost';
```
Switch to the apiservice database:
```
MariaDB [(none)]> USE apiservice;
```

Create the link table:

MariaDB [(apiservice)]> CREATE TABLE IF NOT EXISTS link (link_id CHAR(36) NOT NULL, link TEXT, PRIMARY KEY (link_id));

View its details with:

DESCRIBE link;

Create the tag table:

MariaDB [(apiservice)]> CREATE TABLE IF NOT EXISTS tag (tag_id CHAR(36) NOT NULL, tag VARCHAR(255), PRIMARY KEY (tag_id), INDEX (tag));

View its details with:

DESCRIBE tag;

Create the taglink table:

MariaDB [(apiservice)]> CREATE TABLE IF NOT EXISTS taglink (tag_id CHAR(36), link_id CHAR(36), CONSTRAINT FOREIGN KEY (tag_id) REFERENCES tag (tag_id), CONSTRAINT FOREIGN KEY (link_id) REFERENCES link (link_id), PRIMARY KEY (tag_id, link_id), INDEX (tag_id), INDEX (link_id));

View its details with:

MariaDB [(apiservice)]> DESCRIBE taglink;

Step 3: Add Data

In order that our GET endpoints will work, we need to populate the database with some data.

From the MariaDB connection, run:

MariaDB [(apiservice)]> INSERT INTO link (link_id, link) VALUES(UUID(), "https://google.com");

MariaDB [(apiservice)]> INSERT INTO link (link_id, link) VALUES(UUID(), "https://amazon.com");

MariaDB [(apiservice)]> INSERT INTO tag (tag_id, tag) VALUES(UUID(), "search");

MariaDB [(apiservice)]> INSERT INTO tag (tag_id, tag) VALUES(UUID(), "shop");

MariaDB [apiservice]> INSERT INTO taglink (tag_id,link_id) VALUES((SELECT tag_id FROM tag WHERE tag="search"),(SELECT link_id FROM link WHERE link="https://google.com"));

MariaDB [apiservice]> INSERT INTO taglink (tag_id,link_id) VALUES((SELECT tag_id FROM tag WHERE tag="shop"),(SELECT link_id FROM link WHERE link="https://amazon.com"));

MariaDB [apiservice]> SELECT * FROM link;

MariaDB [apiservice]> SELECT * FROM tag;

MariaDB [apiservice]> SELECT * FROM taglink;

Take a note of the link_id and tag_id values returned from the last query. You’ll need them later. Open a new terminal window, and keep the terminal open. Take the tag_id and link_id from the first row and assign them to environment vars TAG_ID1 and LINK_ID1. Then take the tag_id and link_id from the second row and assign them to TAG_ID2 and LINK_ID2:

$ export $TAG_ID1=xxxx
$ export $LINK_ID1=xxxx
$ export $TAG_ID2=xxxx
$ export $LINK_ID2=xxxx

Step 4: Install MariaDB Connectors

Now we need to install the bits that connect the MariaDB database to the python code.

Open another SSH terminal, and run the following to get the MariaDB repo setup scripts:
```
$ wget https://downloads.mariadb.com/MariaDB/mariadb_repo_setup 
```

Then run the setup:

$ chmod +x mariadb_repo_setup 
$ sudo ./mariadb_repo_setup --mariadb-server-version="mariadb-10.6"

Next install the required packages:

$ sudo yum install MariaDB-shared MariaDB-devel python3-devel gcc

Step 5: Update the Python Virtual Environment

Next we need to add required packages to our FastAPI python venv.

Activate the FastAPI venv:
```
$ . ~/.venv-fastapi/bin/activate
```

Now install the sqlalchemy and mariadb packages:

$ (.venv-fastapi) pip install sqlalchemy
$ (.venv-fastapi) pip install mariadb

Step 6: Update Python Code

Next we need to update the python API code to talk to the database.

Change to the code directory (~/vboxshare/fastapi should be replaced with the path to your FastAPI python code):
```
$ cd ~/vboxshare/fastapi
```
Create a file called config.yaml and add the following to it. Replace PASSWORD with the password you chose when creating the apiservice_user database user:
```
# database config
database:
  username: apiservice_user
  password: 'PASSWORD'
  host: 127.0.0.1
  name: apiservice
  port: 3306
```
Create a new directory called ‘manager’ and change directory to it:
```
$ mkdir -p manager
$ cd manager
```
Create a new empty file called __init__.py
```
$ touch __init__.py
```

Create a new file called database.py and add the following code to it:

from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
import yaml

with open("config.yaml") as file_handle:
    config = yaml.safe_load(file_handle)

database_config = config['database']
username = database_config['username']
password = database_config['password']
host = database_config['host']
name = database_config['name']
port = database_config['port']

SQLALCHEMY_DATABASE_URL = f"mariadb+mariadbconnector://{username}:{password}@{host}:{port}/{name}"

engine = create_engine(SQLALCHEMY_DATABASE_URL)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

Base = declarative_base()

Create a new file called manager.py and add the following code to it:

from typing import Optional

from sqlalchemy.orm import Session

from manager import models


def get_link(db: Session, link_id: str):
    return db.query(models.Link).filter(models.Link.link_id == link_id).first()


def get_tag(db: Session, tag_id: str):
    return db.query(models.Tag).filter(models.Tag.tag_id == tag_id).first()


def get_tag_by_tag_name(db: Session, tag: str):
    return db.query(models.Tag).filter(models.Tag.tag == tag).first()


def get_links(db: Session, tag_id: Optional[str] = None, tag: Optional[str] = None):
    if tag is None and tag_id is None:
        # TODO: Implement offset and limit
        return db.query(models.Link).all()
    elif tag is not None:
        tag_record = get_tag_by_tag_name(db, tag)
        if tag_record is None:
            return []
        tag_id = tag_record.tag_id

    return db.query(models.Link).join(models.TagLink).filter(models.TagLink.tag_id == tag_id).all()


def get_tags(db: Session, tag: Optional[str] = None):
    if tag is None:
        # TODO: Implement offset and limit
        return db.query(models.Tag).all()
    else:
        return get_tag_by_tag_name(db, tag)


def get_taglinks(db: Session, tag_id: Optional[str] = None, link_id: Optional[str] = None):
    if tag_id is None and link_id is None:
        # TODO: Implement offset and limit
        return db.query(models.TagLink).all()
    elif tag_id is not None and link_id is None:
        return db.query(models.TagLink).filter(models.TagLink.tag_id == tag_id).all()
    elif tag_id is None and link_id is not None:
        return db.query(models.TagLink).filter(models.TagLink.link_id == link_id).all()
    else:
        return db.query(models.TagLink).filter(models.TagLink.link_id == link_id, models.TagLink.tag_id == tag_id).all()

Create a new file called models.py and add the following code to it:

from sqlalchemy import Column, ForeignKey, String, UniqueConstraint

from manager.database import Base


class Link(Base):
    __tablename__ = "link"

    link_id = Column(String, primary_key=True, index=True)
    link = Column(String)


class Tag(Base):
    __tablename__ = "tag"

    tag_id = Column(String, primary_key=True, index=True)
    tag = Column(String)


class TagLink(Base):
    __tablename__ = "taglink"

    tag_id = Column(String, ForeignKey("tag.tag_id"), primary_key=True, index=True)
    link_id = Column(String, ForeignKey("link.link_id"), primary_key=True, index=True)

Change directory back to the top level directory

$ cd ../

Replace the contents of main.py with the following:

from typing import Optional

from sqlalchemy.orm import Session

from fastapi import FastAPI, HTTPException, Depends

from manager import manager

from manager.database import SessionLocal


# Dependency
def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()


app = FastAPI()


# Get link by link_id
@app.get("/link/{link_id}")
async def get_link(link_id: str, db: Session = Depends(get_db)):
    db_link = manager.get_link(db, link_id)
    if db_link is None:
        raise HTTPException(status_code=404, detail="Link not found")
    return db_link


# Get links by query params
@app.get("/link/")
async def get_links(tag_id: Optional[str] = None, tag: Optional[str] = None, db: Session = Depends(get_db)):
    return manager.get_links(db, tag_id, tag)


# Get tag by tag_id
@app.get("/tag/{tag_id}")
async def get_tag(tag_id: str, db: Session = Depends(get_db)):
    db_tag = manager.get_tag(db, tag_id)
    if db_tag is None:
        raise HTTPException(status_code=404, detail="Tag not found")
    return db_tag


# Get tags by query params
@app.get("/tag/")
async def get_tags(tag: Optional[str] = None, db: Session = Depends(get_db)):
    return manager.get_tags(db, tag)


# Get taglinks by query params
@app.get("/taglink/")
async def get_taglinks(link_id: Optional[str] = None, tag_id: Optional[str] = None,  db: Session = Depends(get_db)):
    return manager.get_taglinks(db, tag_id, link_id)

Step 7: Test the API

Now we need to get the API running and make some test GET requests.

Start the FastAPI service from the directory where the main.py file is located:

$ (.venv-fastapi) cd ~/vboxshare/fastapi
$ (.venv-fastapi) uvicorn --host 0.0.0.0 main:app --reload

In the SSH terminal where you ran your ‘export’ statements, install jq, a useful json parser tool:
```
$ sudo yum install jq
```

Create a new file called test.sh, and enter the following code:

IP=$1

echo "GET /link link_id=$LINK_ID1"
curl -X 'GET' "http://$IP:8000/link/?link_id=$LINK_ID1" -H 'accept: application/json' | jq .
echo "================"
echo "GET /link tag_id=$TAG_ID2"
curl -X 'GET' "http://$IP:8000/link/?tag_id=$TAG_ID2" -H 'accept: application/json' | jq .
echo "================"
echo "GET /link"
curl -X 'GET' "http://$IP:8000/link/" -H 'accept: application/json' | jq .
echo "================"
echo "GET /link, tag=shop"
curl -X 'GET' "http://$IP:8000/link/?tag=shop" -H 'accept: application/json' | jq .
echo "================"
echo "GET /tag"
curl -X 'GET' "http://$IP:8000/tag/" -H 'accept: application/json' | jq .
echo "================"
echo "GET /tag tag=search"
curl -X 'GET' "http://$IP:8000/tag/?tag=search" -H 'accept: application/json' | jq .
echo "================"
echo "GET /tag tag_id=$TAG_ID1"
curl -X 'GET' "http://$IP:8000/tag/?tag_id=$TAG_ID1" -H 'accept: application/json' | jq .
echo "================"
echo "GET /taglink"
curl -X 'GET' "http://$IP:8000/taglink/" -H 'accept: application/json' | jq .
echo "================"
echo "GET /taglink link_id=$LINK_ID2"
curl -X 'GET' "http://$IP:8000/taglink/?link_id=$LINK_ID2" -H 'accept: application/json' | jq .
echo "================"
echo "GET /taglink tag_id=$TAG_ID1"
curl -X 'GET' "http://$IP:8000/taglink/?tag_id=$TAG_ID1" -H 'accept: application/json' | jq .
echo "================"
echo "GET /taglink link_id=$LINK_ID1, tag_id=$TAG_ID1"
curl -X 'GET' "http://$IP:8000/taglink/?link_id=$LINK_ID1&tag_id=$TAG_ID1" -H 'accept: application/json' | jq .
echo "================"

Make sure the test script is executable:
```
$ chmod ugo+x test.sh
```
Run the test script, replacing YOUR_IP with the IP of your API host server (or 127.0.0.1 if running the test script from the machine that is hosting the API):
```
$ ./test.sh YOUR_IP
```
If all goes well, you should see an output of json responses to each request, with no errors.

Conclusion

You should now have a FastAPI API running with five GET endpoints, that call into a MariaDB database to retrieve data.

If you want to find out how to add POST endpoints to the API, see my follow-on post, Creating an API with python: Part 3: POST Endpoints.

Thanks for reading!

Categories

About Liz

Subscribe

Recent Posts

Recent Comments

Creating an API with python: Part 2: MariaDB Database

Prerequisites

Step 1: Install MariaDB

Step 2: Set Up Database Users and Tables

Step 3: Add Data

Step 4: Install MariaDB Connectors

Step 5: Update the Python Virtual Environment

Step 6: Update Python Code

Step 7: Test the API

Conclusion

You may also like...

1 Response

Leave a Reply Cancel reply

Prerequisites

Step 1: Install MariaDB

Step 2: Set Up Database Users and Tables

Step 3: Add Data

Step 4: Install MariaDB Connectors

Step 5: Update the Python Virtual Environment

Step 6: Update Python Code

Step 7: Test the API

Conclusion

You may also like...

Sending Emails with Python

Creating a Crypto Stats Page: Part 2: Asset Pair Dropdown

You Already Know How to Code: Part 2

1 Response

Leave a Reply Cancel reply