FastAPI project - CRUD operations on database

Question

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

• Martin R: I have rolled back your last edit. Please don't change or add to the code in your question after you have received answers. See What should I do when someone answers my question? Thank you.

Answer 1

I'm not intimately familiar with FastAPI, but

raise HTTPException(status_code=status.HTTP_200_OK,
 detail="Welcome to PokeAPI")

is surely not the best way to return a string with a success code. Exceptions are exceptional. You should probably be using a Response and setting its status_code field, something like in here: https://fastapi.tiangolo.com/advanced/response-change-status-code/

get_poke_by_name appears to return a tuple with fields by position. This really should be avoided, and you should have a lightweight class like a NamedTuple returned instead - or better yet, just use Row which supports named access. This is especially important given that you've used select * which makes no guarantees about column order.

conn and cursor need to be protected in context management with statements.

Add PEP484 type hints to your function signatures, particularly for methods like add_poke_to_db.

• Eliran Turgeman: I changed everything according to your review apart from the NamedTuple thing because I just wasn't sure how to do it (if you could guide me a bit that would be very appreciated). I just specified in the query all the columns I want to select instead of using SELECT * - Is it a good solution too? You can see the changes I made to the code in the edit I made to the question
• Reinderien: Skip the named tuple for now and pursue the documentation I linked for Row, which will be assigned to the row_factory member