Creating an API with python: Part 3: POST Endpoints

In my previous post, Creating an API with python: Part 2: MariaDB Database, I set up a MariaDB database and connected it to the five GET endpoints in the FastAPI API. In this post, I’ll add three POST endpoints which will add data to the database.

Prerequisites

Step 1: Add schemas.py

Create a new python file called schemas.py. This will contain the pydantic schemas that will be used to validate the POST data for the POST endpoints. They will also provide descriptions for the swagger docs page.

Change to the code directory (~/vboxshare/fastapi should be replaced with the path to your FastAPI python code):
```
$ cd ~/vboxshare/fastapi
```
Change to the manager directory.
```
$ cd manager
```

Create a file called schemas.py and add the following to it:

from typing import Optional

from pydantic import BaseModel, Field


class PostLink(BaseModel):
    link: str = Field(..., description="The link URL")
    tag: Optional[str] = Field(None,
                               description="Tag name to associate with the link (will be created if it doesn't exist)")
    tag_id: Optional[str] = Field(None, description="Tag ID to associate with the link (must already exist)")


class PostTag(BaseModel):
    tag: str = Field(..., description="Tag name")


class PostTagLink(BaseModel):
    tag_id: str = Field(..., description="Tag ID (must already exist)")
    link_id: str = Field(..., description="Link ID (must already exist)")

Step 2: Update manager.py

Update the manager/manager.py file with new functions for creating a link, tag and taglink (create_link, create_tag, create_taglink).

Open the manager/manager.py file and replace the contents with the following code:

from typing import Optional

from uuid import uuid4

from sqlalchemy.orm import Session

from fastapi import HTTPException

from manager import models, schemas


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:
        db_tag = get_tag_by_tag_name(db, tag)
        if db_tag is not None:
            return [db_tag]
        return []


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


def create_link(db: Session, link: schemas.PostLink):
    db_link = models.Link(link_id=str(uuid4()), link=link.link)
    db.add(db_link)
    tag_id = link.tag_id
    link_id = db_link.link_id

    if link.tag is not None:
        db_tag = get_tag_by_tag_name(db, link.tag)
        if db_tag is None:
            db_tag = models.Tag(tag_id=str(uuid4()), tag=link.tag)
            db.add(db_tag)
            tag_id = db_tag.tag_id
        else:
            tag_id = db_tag.tag_id
    elif tag_id is not None:
        db_tag = get_tag(db, tag_id)
        if db_tag is None:
            raise HTTPException(status_code=404, detail=f"Tag with tag_id {tag_id} not found")

    db_taglink = models.TagLink(link_id=link_id, tag_id=tag_id)
    db.add(db_taglink)

    db.commit()
    db.refresh(db_link)
    return db_link


def create_tag(db: Session, tag: schemas.PostTag):
    db_tag = models.Tag(tag_id=str(uuid4()), tag=tag.tag)
    db_tag_existing = get_tags(db, tag=tag.tag)
    if len(db_tag_existing) > 0:
        raise HTTPException(status_code=409, detail=f"Tag with name {tag.tag} exists")
    db.add(db_tag)
    db.commit()
    db.refresh(db_tag)
    return db_tag


def create_taglink(db: Session, taglink: schemas.PostTagLink):
    tag_id = taglink.tag_id
    link_id = taglink.link_id
    db_taglink = models.TagLink(tag_id=tag_id, link_id=link_id)
    db_tag_id_existing = get_tag(db, tag_id)
    if db_tag_id_existing is None:
        raise HTTPException(status_code=422, detail=f"Tag with tag_id {tag_id} not found")
    db_link_id_existing = get_link(db, link_id)
    if db_link_id_existing is None:
        raise HTTPException(status_code=422, detail=f"Link with link_id {link_id} not found")
    db_taglink_existing = get_taglinks(db, tag_id=tag_id, link_id=link_id)
    if len(db_taglink_existing) > 0:
        raise HTTPException(status_code=409, detail=f"TagLink with tag_id {tag_id} and link_id {link_id} exists")
    db.add(db_taglink)
    db.commit()
    db.refresh(db_taglink)
    return db_taglink

Step 3: Update main.py

Update main.py with a rollback in the event of a database exception, and add functions post_link, post_tag and post_taglink as the three new POST endpoint functions.

Change directory to the top level directory:
```
$ cd ../
```

Replace the contents of main.py with:

from typing import Optional

from sqlalchemy.orm import Session

from fastapi import FastAPI, HTTPException, Depends

from manager import manager, schemas

from manager.database import SessionLocal


# Dependency
def get_db():
    db = SessionLocal()
    try:
        yield db
    except Exception:
        db.rollback()
    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)


# Post a link
@app.post("/link/")
async def post_link(link: schemas.PostLink, db: Session = Depends(get_db)):
    if link.tag is None and link.tag_id is None:
        raise HTTPException(status_code=422, detail="One of tag_id or tag must be specified")

    if link.tag is not None and link.tag_id is not None:
        raise HTTPException(status_code=422, detail="Only one of tag_id or tag must be specified")

    db_link = manager.create_link(db, link)

    return db_link


# Post a tag
@app.post("/tag/")
async def post_tag(tag: schemas.PostTag, db: Session = Depends(get_db)):
    db_tag = manager.create_tag(db, tag)

    return db_tag


# Post a taglink
@app.post("/taglink/")
async def post_taglink(taglink: schemas.PostTagLink, db: Session = Depends(get_db)):
    db_tag = manager.create_taglink(db, taglink)

    return db_tag

Step 4: Update test script

Now we need to update the test script to test the POST endpoints, retrieve the responses and test the GET endpoints with the data retrieved.

Open the file test.sh, and replace the code with the following:

IP=$1
BASEURL=http://$IP:8000

echo "=========================="
echo "Test POST /link link=https://www.test1.com, tag=test1"
resp=$(curl -X "POST" -H "Content-Type: application/json" -d "{\"link\": \"https://www.test1.com\", \"tag\": \"test1\"}" $BASEURL/link/)
echo $resp
link_id1=$(echo $resp | jq -r '.link_id')
echo $link_id1

echo "=========================="
echo "Test GET /tag tag=test1"
resp=$(curl -X "GET" -H "Content-Type: application/json" $BASEURL/tag/?tag=test1)
echo $resp
tag_id1=$(echo $resp | jq -r '.[0].tag_id')
echo $tag_id1

echo "=========================="
echo "Test POST /tag tag=test2"
resp=$(curl -X "POST" -H "Content-Type: application/json" -d "{\"tag\": \"test2\"}" $BASEURL/tag/)
echo $resp
tag_id2=$(echo $resp | jq -r '.tag_id')
echo $tag_id2

echo "=========================="
echo "Test POST /link link=https://www.test2.com, tag=test2"
resp=$(curl -X "POST" -H "Content-Type: application/json" -d "{\"link\": \"https://www.test2.com\", \"tag\": \"test2\"}" $BASEURL/link/)
echo $resp
link_id2=$(echo $resp | jq -r '.link_id')
echo $link_id2

echo "=========================="
echo "Test POST /link link=https://www.test2.com, tag_id=$tag_id1"
resp=$(curl -X "POST" -H "Content-Type: application/json" -d "{\"link\": \"https://www.test2.com\", \"tag_id\": \"$tag_id1\"}" $BASEURL/link/)
echo $resp
link_id2=$(echo $resp | jq -r '.link_id')
echo $link_id2

echo "=========================="
echo "Test POST /link link=https://www.test2.com, tag_id=invalid. Expect 404 response: Tag with tag_id invalid not found"
resp=$(curl -X "POST" -H "Content-Type: application/json" -d "{\"link\": \"https://www.test2.com\", \"tag_id\": \"invalid\"}" $BASEURL/link/)
echo $resp

echo "=========================="
echo "Test GET /link"
resp=$(curl -X "GET" -H "Content-Type: application/json" $BASEURL/link/)
echo $resp

echo "=========================="
echo "Test GET /link tag_id=$tag_id1"
resp=$(curl -X "GET" -H "Content-Type: application/json" $BASEURL/link/?tag_id=$tag_id1)
echo $resp

echo "=========================="
echo "Test GET /link tag=test1"
resp=$(curl -X "GET" -H "Content-Type: application/json" $BASEURL/link/?tag=test1)
echo $resp

echo "=========================="
echo "Test GET /link link_id=$link_id1"
resp=$(curl -X "GET" -H "Content-Type: application/json" $BASEURL/link/$link_id1)
echo $resp

echo "=========================="
echo "Test GET /tag tag_id=$tag_id1"
resp=$(curl -X "GET" -H "Content-Type: application/json" $BASEURL/tag/$tag_id1)
echo $resp

echo "=========================="
echo "Test GET /tag"
resp=$(curl -X "GET" -H "Content-Type: application/json" $BASEURL/tag/)
echo $resp

echo "=========================="
echo "Test POST /taglink tag_id=$tag_id2, link_id=$link_id1"
resp=$(curl -X "POST" -H "Content-Type: application/json" -d "{\"link_id\": \"$link_id1\", \"tag_id\": \"$tag_id2\"}" $BASEURL/taglink/)
echo $resp

echo "=========================="
echo "Test POST /taglink tag_id=invalid, link_id=$link_id1. Expect 422 response: Tag with tag_id invalid not found"
resp=$(curl -X "POST" -H "Content-Type: application/json" -d "{\"link_id\": \"$link_id1\", \"tag_id\": \"invalid\"}" $BASEURL/taglink/)
echo $resp

echo "=========================="
echo "Test POST /taglink tag_id=$tag_id1, link_id=invalid. Expect 422 response: Link with link_id invalid not found"
resp=$(curl -X "POST" -H "Content-Type: application/json" -d "{\"link_id\": \"invalid\", \"tag_id\": \"$tag_id1\"}" $BASEURL/taglink/)
echo $resp

echo "=========================="
echo "Test POST /taglink tag_id=$tag_id1, link_id=$link_id1. Expect 409 response: TagLink with tag_id $tag_id1 and link_id $link_id1 exists"
resp=$(curl -X "POST" -H "Content-Type: application/json" -d "{\"link_id\": \"$link_id1\", \"tag_id\": \"$tag_id1\"}" $BASEURL/taglink/)
echo $resp

echo "=========================="
echo "Test GET /taglink"
resp=$(curl -X "GET" -H "Content-Type: application/json" $BASEURL/taglink/)
echo $resp

echo "=========================="
echo "Test GET /taglink link_id=$link_id1"
resp=$(curl -X "GET" -H "Content-Type: application/json" $BASEURL/taglink/?link_id=$link_id1)
echo $resp

echo "=========================="
echo "Test GET /taglink tag_id=$tag_id1"
resp=$(curl -X "GET" -H "Content-Type: application/json" $BASEURL/taglink/?tag_id=$tag_id1)
echo $resp

echo "=========================="
echo "Test GET /taglink tag_id=$tag_id1 link_id=$link_id1"
resp=$(curl -X "GET" -H "Content-Type: application/json" $BASEURL/taglink/?tag_id=$tag_id1&link_id=$link_id1)
echo $resp

Make sure the test script is executable:
```
$ chmod ugo+x test.sh
```

Step 5: Run the test script

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, other than those expected in the test descriptions.
If you want to run the test script again, you will need to clear the test data from the database first. To do this, open a mysql terminal:
```
$ mysql -u root -p
```
Enter the password you set with the initial set-up of MariaDB in Part 2.
Switch to the apiservice database:
```
MariaDB [(none)]> USE apiservice;
```

Run the following to clear the test data:

MariaDB [(apiservice)]> DELETE from taglink where link_id IN(SELECT link_id from link where link LIKE '%test%'); DELETE from taglink where tag_id IN(SELECT tag_id from tag where tag LIKE '%test%'); DELETE from tag where tag LIKE '%test%'; DELETE from link where link LIKE '%test%';

You should now be able to run the test script again.

Conclusion

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

To add DELETE endpoints, see my follow-on post, Creating an API with python: part 4: DELETE Endpoints.

Thanks for reading!

Categories

About Liz

Subscribe

Recent Posts

Recent Comments

Creating an API with python: Part 3: POST Endpoints

Prerequisites

Step 1: Add schemas.py

Step 2: Update manager.py

Step 3: Update main.py

Step 4: Update test script

Step 5: Run the test script

Conclusion

You may also like...

1 Response

Leave a Reply Cancel reply

Prerequisites

Step 1: Add schemas.py

Step 2: Update manager.py

Step 3: Update main.py

Step 4: Update test script

Step 5: Run the test script

Conclusion

You may also like...

Creating a Python Decorator

Why it’s not Wise to use a Mutable Default

You Already Know How to Code: Part 5

1 Response

Leave a Reply Cancel reply