Пример FastAPI Python

Введение

Пример создания API на основе FastAPI и SQLite3

Схема

В данном примере я запускаю FastAPI облачном сервере Beget но на локальном хосте всё должно работать так же.

├── db │ ├── create_all_testers_table.py │ ├── create_tools_table.py │ ├── list_all_tables.py │ └── testers.db ├── main.py ├── requirements.txt └── restart.sh

Код

# main.py import sqlite3 from fastapi import FastAPI, Response, status from pydantic import BaseModel app = FastAPI() class Factors(BaseModel): first: float second: float model_config = { "json_schema_extra": { "examples": [ { "first": 2, "second": 5 } ] } } class Tester(BaseModel): name: str tool: str testers_list = [ "Yuliya", "Dusan", "Konstantin", "Julia", "Arya", "Andrei" ] @app.get("/", description="This is our first route.") async def root(): return {"message": "Thank you for visiting JustAPI.ru" } @app.post("/") async def post(): return {"message": "hello from the post route"} @app.put("/") async def put(): return {"message": "hello from the put route"} @app.get("/testers") async def testers() -> list: return testers_list @app.put("/multiply") async def multiply(factor_1: float, factor_2: float): return factor_1 * factor_2 @app.post("/multiply") async def multiply(f: Factors): return f.first * f.second @app.post("/testers/add", status_code=200) async def add_tester(tester: Tester): print(tester.name) con = sqlite3.connect('db/testers.db') cur = con.cursor() sql = f""" INSERT INTO all_testers (name, tool) VALUES (?,?) """ try: cur.execute(sql, [tester.name, tester.tool]) except: return Response(status_code=status.HTTP_412_PRECONDITION_FAILED) con.commit() con.close() return Response(status_code=status.HTTP_200_OK) @app.get("/testers/list") async def list_testers() -> list: testers_list = [] con = sqlite3.connect('db/testers.db') cur = con.cursor() sql = f""" SELECT name FROM all_testers """ cur.execute(sql) names = cur.fetchall() for name in names: print(name) testers_list.append(name[0]) con.commit() con.close() return testers_list

# create_all_testers_table.py import sqlite3 con = sqlite3.connect('testers.db') cur = con.cursor() cur.execute("CREATE TABLE all_testers (tester_id INTEGER PRIMARY KEY AUTOINCREMENT, " "name TEXT NOT NULL UNIQUE, tool TEXT);") con.commit() con.close()

# create_tools_table.py import sqlite3 con = sqlite3.connect('testers.db') cur = con.cursor() cur.execute("CREATE TABLE tools (tool_id, name, users, rating);") con.commit() con.close()

# list_all_tables.py import sqlite3 try: # Making a connection between sqlite3 # database and Python Program sqliteConnection = sqlite3.connect('testers.db') # If sqlite3 makes a connection with python # program then it will print "Connected to SQLite" # Otherwise it will show errors print("Connected to testers.db") # Getting all tables from sqlite_master sql_query = """SELECT name FROM sqlite_master WHERE type='table';""" # Creating cursor object using connection object cursor = sqliteConnection.cursor() # executing our sql query cursor.execute(sql_query) print("List of tables\n") tables = cursor.fetchall() # printing all tables list # print(tables) # [('all_testers',), ('tools',)] for table in tables: # print(table) table_name = table[0] print(f"\nTable: {table_name}\n") sql_query = f"SELECT * from {table_name}" cursor = sqliteConnection.execute(sql_query) names = list(map(lambda x: x[0], cursor.description)) for name in names: print(f"column: {name}") except sqlite3.Error as error: print("Failed to execute the above query", error) finally: # Inside Finally Block, If connection is # open, we need to close it if sqliteConnection: # using close() method, we will close # the connection sqliteConnection.close() # After closing connection object, we # will print "the sqlite connection is # closed" print("the sqlite connection is closed")

Автор статьи: Андрей Олегович

РЕКЛАМА хостинга Beget, которым я пользуюсь более десяти лет

Конец рекламы хостинга Beget, который я всем рекомендую.