python – FastAPI project – CRUD operations on database

recently I have been trying to educate myself on REST APIs.

As a part of that, I have created a small project using FastAPI in python and I would love to hear some feedback about it.

When I wrote this, I didn’t take into consideration any design pattern, and honestly, I almost knew nothing about them – on the last week I have been trying to educate myself on that subject as well, and I would appreciate some advice on what design pattern should I use for this API? ( I think Pagination is quite fitting ).

The code purpose is –

  • query database for pokemons information by parameters such as name and type
  • Create a new pokemon (add to the database)
  • Delete Pokemon from the database based on name
  • Update pokemon information based on name

This project relies on two main files:

pokeapi.py:

from typing import Optional

from fastapi import FastAPI, Path, HTTPException, status
from pydantic import BaseModel

from database import get_poke_by_name, get_poke_by_type, add_poke_to_db, 
    update_poke, delete_poke

app = FastAPI()


class Pokemon(BaseModel):
    name: str
    primary_type: str
    secondary_type: str
    sum_stats: int
    hit_points: int
    attack_strength: int
    defensive_strength: int
    special_attack_strength: int
    special_defensive_strength: int


@app.get("/")
def root():
    raise HTTPException(status_code=status.HTTP_200_OK,
                        detail="Welcome to PokeAPI")


@app.get("/poke/{pokemon_name}")
def get_pokemon_by_name(pokemon_name: str = Path(None,
                                                 description="Name of the "
                                                             "pokemon you'd "
                                                             "like to "
                                                             "retrieve")):
    pokemon = get_poke_by_name(pokemon_name)
    if not pokemon:
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND,
                            detail="Pokemon not found")

    return {"Pokemon": pokemon(0),
            "Types": (pokemon(1), pokemon(2)),
            "HP": pokemon(4),
            "Attack": pokemon(5),
            "Special Attack": pokemon(6),
            "Defense": pokemon(7),
            "Special Defense": pokemon(8),
            }


@app.get("/poketype/{poke_type}")
def get_pokemon_by_type(poke_type: str = Path(None,
                                              description="Primary type of "
                                                          "the pokemons you "
                                                          "want to query"),
                        type2: Optional(str) = None):
    pokemons = get_poke_by_type(poke_type, type2)
    if not pokemons:
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND,
                            detail="No pokemon with this type")
    result = {}
    for idx, pokemon in enumerate(pokemons):
        result(idx) = {"Pokemon": pokemon(0),
                       "Types": (pokemon(1), pokemon(2)),
                       "HP": pokemon(4),
                       "Attack": pokemon(5),
                       "Special Attack": pokemon(6),
                       "Defense": pokemon(7),
                       "Special Defense": pokemon(8),
                       }

    return result


@app.post("/newPoke/{pokemon_name}")
def create_pokemon(pokemon_name: str, pokemon: Pokemon):
    if get_poke_by_name(pokemon_name):
        raise HTTPException(status_code=status.HTTP_406_NOT_ACCEPTABLE,
                            detail="Pokemon already exists")
    add_poke_to_db(pokemon.name, pokemon.primary_type, pokemon.secondary_type,
                   pokemon.sum_stats, pokemon.hit_points,
                   pokemon.attack_strength, pokemon.special_attack_strength,
                   pokemon.defensive_strength,
                   pokemon.special_defensive_strength)
    raise HTTPException(status_code=status.HTTP_201_CREATED,
                        detail="Pokemon created successfully")


@app.put("/updatePoke/{pokemon_name}")
def update_pokemon(pokemon_name: str, pokemon: Pokemon):
    if not get_poke_by_name(pokemon_name):
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND,
                            detail="Pokemon not found")

    update_poke(pokemon.name, pokemon.primary_type, pokemon.secondary_type,
                pokemon.sum_stats, pokemon.hit_points,
                pokemon.attack_strength, pokemon.special_attack_strength,
                pokemon.defensive_strength,
                pokemon.special_defensive_strength)

    raise HTTPException(status_code=status.HTTP_200_OK,
                        detail="Pokemon details updated")


@app.delete("/deletePoke/{pokemon_name}")
def delete_pokemon(pokemon_name: str):
    if not get_poke_by_name(pokemon_name):
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND,
                            detail="Pokemon not found")

    delete_poke(pokemon_name)

    raise HTTPException(status_code=status.HTTP_200_OK,
                        detail="Pokemon deleted successfully")

and database.py

from pathlib import Path
import sqlite3
import pandas as pd

DB_FILENAME = "poke_db.db"


def table_exists(cursor):
    cursor.execute('''
        SELECT count(name) FROM sqlite_master WHERE type='table' AND name='Pokemons' ''')

    if not cursor.fetchone()(0):
        return False

    return True


def init_db():
    if not Path(DB_FILENAME).is_file():
        Path(DB_FILENAME).touch()


def load_csv_to_db():
    init_db()
    conn = sqlite3.connect(DB_FILENAME)
    cursor = conn.cursor()
    cursor.execute('''
        CREATE TABLE IF NOT EXISTS Pokemons (name text, type1 text,
        type2 text, sum_stats int, hp int, attack int,
        special_attack int, defense int, special_defense int)
        ''')

    poke_data = pd.read_csv('Pokemon.csv')
    poke_data.drop(('#', 'Speed', 'Generation', 'Legendary'), axis=1, inplace=True)

    poke_data.columns = ('name', 'type1', 'type2', 'sum_stats',
                         'hp', 'attack', 'special_attack', 'defense',
                         'special_defense')

    poke_data.to_sql('Pokemons', conn, if_exists='append', index=False)


def get_poke_by_name(poke_name):
    conn = sqlite3.connect(DB_FILENAME)
    cursor = conn.cursor()

    if not table_exists(cursor):
        load_csv_to_db()

    cursor.execute('''SELECT * FROM Pokemons WHERE name = ?''', (poke_name,))
    return cursor.fetchone()


def get_poke_by_type(type1, type2):
    conn = sqlite3.connect(DB_FILENAME)
    cursor = conn.cursor()

    if not table_exists(cursor):
        load_csv_to_db()

    if type2:
        cursor.execute('''
        SELECT * FROM Pokemons WHERE type1 = ? AND type2 = ?''', (type1, type2))

    else:
        cursor.execute('''
        SELECT * FROM Pokemons WHERE type1 = ?''', (type1,))

    return cursor.fetchall()


def add_poke_to_db(name, type1, type2, sum_stats, hp, attack, special_attack,
                   defense, special_defense):
    conn = sqlite3.connect(DB_FILENAME)
    cursor = conn.cursor()

    if not table_exists(cursor):
        load_csv_to_db()

    cursor.execute('''
    INSERT INTO Pokemons ('name', 'type1', 'type2', 'sum_stats',
    'hp', 'attack', 'special_attack', 'defense', 'special_defense')
    VALUES (?,?,?,?,?,?,?,?,?)''', (name, type1, type2, sum_stats, hp, attack,
                                    special_attack, defense, special_defense))

    conn.commit()


def update_poke(name, type1=None, type2=None, sum_stats=None, hp=None,
                attack=None, special_attack=None, defense=None,
                special_defense=None):

    conn = sqlite3.connect(DB_FILENAME)
    cursor = conn.cursor()

    if not table_exists(cursor):
        load_csv_to_db()

    params = (type1, type2, sum_stats, hp, attack, special_attack,
              defense, special_defense)
    params_names = ('type1', 'type2', 'sum_stats', 'hp', 'attack',
                    'special_attack', 'defense', 'special_defense')

    for param, param_name in zip(params, params_names):
        if param:
            query = '''
            UPDATE Pokemons SET ''' + param_name + '''
             = ? WHERE name = ?'''
            cursor.execute(query, (param, name))

    conn.commit()


def delete_poke(name):
    conn = sqlite3.connect(DB_FILENAME)
    cursor = conn.cursor()

    if not table_exists(cursor):
        load_csv_to_db()

    cursor.execute('''
    DELETE FROM Pokemons WHERE name = ?''', (name,))

    conn.commit()

If it will be more comfortable, here’s a link to the GitHub repo
https://github.com/DevEliran/PokeAPI

I’d love your review on any small or big detail