В этой статье нам нужно будет построить API, используя Framework API Falcon в Python. Мы собираемся покрыть некоторые намеки и методы для запроса. Эта статья является частью серии:
1) Bootstraphy Falcon API Framework на Docker
2) Создание API с помощью отдельных методов сокола. (Ты здесь.)
Вы лучше прочитали предыдущую статью, или вы будете путать об этом. Правильно, давайте начнем.
Мой корневой каталог является следующим. Прежде всего, я создал требования .txt файл.
attrs==19.1.0 falcon==2.0.0 falcon-autocrud==1.0.36 gunicorn==19.9.0 jsonschema==3.0.1 marshmallow==2.19.5 psycopg2==2.8.3 pyrsistent==0.15.4 python-dateutil==2.8.0 six==1.12.0 SQLAlchemy==1.3.6 webargs==5.4.0
Это готов извлечь библиотеки по этой команде.
pip install -r requirements.txt
После организации библиотек мы можем создать файлы Python. Мой основной файл это app.py Отказ Несмотря на это, вы можете определить другое имя.
import falcon
from resources import athlete
from resources import plan
from resources import exercise
from services import database_service
from middlewares import (
ContentEncodingMiddleware,
)
conn = database_service.connect()
api = falcon.API(middleware=[
ContentEncodingMiddleware(),
])
# api = falcon.API()
athlete = athlete.Athlete(conn, database_service)
plan = plan.Plan(conn, database_service)
exercise = exercise.Exercise(conn, database_service)
api.add_route('/athletes/{id}', athlete)
api.add_route('/athletes', athlete, suffix='collection')
api.add_route('/plans/{id}', plan)
api.add_route('/plans', plan, suffix='collection')
api.add_route('/exercises/{id}', exercise)
api.add_route('/exercises', exercise, suffix='collection')
Этот файл предоставляет приложение Falcon к Bootstrap. Я добавил свои маршруты и связанные файлы в этот файл. Теперь, как вы видите, есть ресурс каталог. Этот каталог отправляет свои объекты. У меня 3 объекта, чтобы обеспечить как ресурсы. Первый – Plan.py Отказ
import falcon
from webargs import fields
from webargs.falconparser import use_args
class Plan(object):
post_request_args = {"name": fields.Str(required=True), "description": fields.Str(required=True),
"difficulty": fields.Int(required=True), "athlete_id": fields.Int(required=True)}
def __init__(self, conn, database_service):
self.conn, self.database_service, self.resource_name = conn, database_service, self.__class__.__name__
def on_delete(self, req, resp, id):
try:
q = " ".join(
["DELETE", "FROM", self.resource_name.lower(), "WHERE", self.resource_name.lower() + "_id = %s"])
q_resp = self.database_service.run_delete_query(self.conn, q, [id])
if not q_resp['status']:
output = {"status": True, "message": q_resp['message'],
"data": None}
else:
output = {"status": True, "message": self.resource_name + " was deleted successfully!", "data": None}
resp.status = falcon.HTTP_200
resp.body = output
except Exception as error:
output = {"status": False, "message": str(error), "data": None}
resp.status = falcon.HTTP_500
resp.body = output
def on_get(self, req, resp, id):
try:
cur = self.conn.cursor()
q = " ".join(
["SELECT", "*", "FROM", self.resource_name.lower(), "wHERE", self.resource_name.lower() + "_id = %s"])
q_resp = self.database_service.run_get_query(cur, q, [id])
if not q_resp['status']:
output = {"status": True, "message": q_resp['message'],
"data": None}
else:
output = {"status": True, "message": None,
'data': self.database_service.set_columns(q_resp['data'], cur)}
resp.status = falcon.HTTP_200
resp.body = output
except Exception as error:
output = {"status": False, "message": str(error), "data": None}
resp.status = falcon.HTTP_500
resp.body = output
def on_get_collection(self, req, resp):
try:
cur = self.conn.cursor()
q = " ".join(
["SELECT * FROM", self.resource_name.lower()])
q_resp = self.database_service.run_get_query(cur, q, [])
if not q_resp['status']:
output = {"status": True, "message": q_resp['message'],
"data": None}
else:
output = {"status": True, "message": None,
'data': self.database_service.set_columns(q_resp['data'], cur)}
resp.status = falcon.HTTP_200
resp.body = output
except Exception as error:
output = {"status": False, "message": str(error), "data": None}
resp.status = falcon.HTTP_500
resp.body = output
def on_put(self, req, resp, id):
try:
cur = self.conn.cursor()
# q = "SELECT name, description, difficulty FROM " + self.resource_name.lower() + " WHERE " + self.resource_name.lower() + "_id = %s;"
get_q = " ".join(
["SELECT name,description,difficulty FROM", self.resource_name.lower(), "wHERE",
self.resource_name.lower() + "_id = %s"])
get_resp = self.database_service.run_get_query(cur, get_q, [id])
record = list(self.database_service.set_columns(get_resp['data'], cur))[0]
request = req.media
for index in record.keys():
if index in request.keys():
record[index] = request[index]
record['id'] = id
update_q = " ".join(
["UPDATE", self.resource_name.lower(), "SET name=%s, description=%s, difficulty=%s WHERE",
self.resource_name.lower() + "_id=%s RETURNING ", self.resource_name.lower() + "_id;"])
update_resp = self.database_service.run_upsert_query(self.conn, update_q, record.values())
if not update_resp['status']:
output = {"status": True, "message": update_resp['message'],
"data": None}
else:
response_data = {
"id": update_resp['data'],
"name": record['name'],
"description": record['description'],
"difficulty": record['difficulty']
}
output = {"status": True, "message": self.resource_name + " is updated successfully!",
"data": response_data}
resp.status = falcon.HTTP_201
resp.body = output
except Exception as error:
output = {"status": False, "message": str(error), "data": None}
resp.status = falcon.HTTP_500
resp.body = output
@use_args(post_request_args)
def on_post_collection(self, req, resp, args):
try:
# q = "INSERT INTO " + self.resource_name.lower() + " (name, description, difficulty, athlete_id) VALUES (%s,%s,%s,%s) RETURNING " + self.resource_name.lower() + "_id;"
q = " ".join(
["INSERT INTO", self.resource_name.lower(),
"(name, description, difficulty, athlete_id) VALUES (%s,%s,%s,%s) RETURNING",
self.resource_name.lower() + "_id;"])
params = {'name': args['name'], 'description': args['description'], 'difficulty': args['difficulty'],
'athlete_id': args['athlete_id']}
q_resp = self.database_service.run_upsert_query(self.conn, q, params.values())
if not q_resp['status']:
output = {"status": True, "message": q_resp['message'],
"data": None}
else:
response_data = {
"id": q_resp['data'],
"name": args['name'],
"description": args['description'],
"difficulty": args['difficulty'],
"athlete_id": args['athlete_id']
}
output = {"status": True, "message": self.resource_name + " is added successfully!",
"data": response_data}
resp.status = falcon.HTTP_201
resp.body = output
except Exception as error:
output = {"status": False, "message": str(error), "data": None}
resp.status = falcon.HTTP_500
resp.body = output
Я собираюсь поделиться только этим файлом в качестве ресурса. Потому что другие файлы точно такие же в соответствии с структурой кода.
Вы увидите префикс перед методами под названием on_ И HTTP глагол приходят после этого префикса. Мы понимаем, каждый метод представляет глаголы HTTP.
Согласно этому, у нас будут эти конечные точки;
1-) http://localhost: 8000/планы (Получить, пост, поставить, удалить)
Совершенство! Это просто для загрузки. Мы можем пройти шаг конфигурации. Поэтому я создал каталог CONF. Он имеет два файла. config.ini и gunicorn_conf.py Отказ INI-файл похож на это.
[postgresqlDB] host = postgresql db = test_dev user = mertingen pass = mertingen
Это файл конфигурации для веб-сервиса.
import multiprocessing bind = '0.0.0.0:8000' workers = multiprocessing.cpu_count() * 2 + 1 timeout = 30 worker_connections = 1000
Я тоже Услуги каталог для обработки некоторых полезных процессов. Это create_schema_service.py
import psycopg2
import database_service
conn = database_service.connect()
def create_tables():
""" create tables in the PostgreSQL database"""
commands = (
"""
CREATE TABLE IF NOT EXISTS athlete (
athlete_id SERIAL PRIMARY KEY,
name VARCHAR(255) NOT NULL,
email VARCHAR(255) NOT NULL,
phone VARCHAR(255) NOT NULL,
gender VARCHAR(255) NOT NULL,
birthday DATE NOT NULL
)
""",
""" CREATE TABLE IF NOT EXISTS plan (
plan_id SERIAL PRIMARY KEY,
athlete_id INTEGER NOT NULL,
name VARCHAR(255) NOT NULL,
description TEXT NOT NULL,
difficulty VARCHAR(255) NOT NULL,
FOREIGN KEY (athlete_id)
REFERENCES athlete (athlete_id)
ON UPDATE CASCADE ON DELETE CASCADE
)
""",
"""
CREATE TABLE IF NOT EXISTS exercise (
exercise_id SERIAL PRIMARY KEY,
name VARCHAR(255) NOT NULL,
description TEXT NOT NULL
)
"""
)
try:
cur = conn.cursor()
for c in commands:
cur.execute(c)
print("Table was created successfully!")
cur.close()
conn.commit()
except (Exception, psycopg2.DatabaseError) as error:
print(error)
finally:
if conn is not None:
conn.close()
if __name__ == '__main__':
create_tables()
И, database_service.py
import psycopg2 as pg
import configparser as cp
import os
dir_path = os.path.dirname(os.path.realpath(__file__))
c = cp.ConfigParser()
c.read(dir_path + '/../conf/config.ini')
def connect():
try:
connection = pg.connect(user=c['postgresqlDB']['user'],
password=c['postgresqlDB']['pass'],
host=c['postgresqlDB']['host'],
port="5432",
database=c['postgresqlDB']['db'])
print("You are connected!")
return connection
except (Exception, pg.Error) as error:
print("Error while connecting to PostgreSQL", error)
# finally:
# if connection:
# connection.close()
# print("PostgreSQL connection is closed")
def set_columns(data, cur):
items = []
if data:
for x in data:
item = {}
c = 0
for col in cur.description:
item.update({col[0]: x[c]})
c = c + 1
items.append(item)
return items
else:
return []
def run_get_query(cur, query, params):
try:
if params:
cur.execute(query, tuple(params))
else:
cur.execute(query)
records = cur.fetchall()
return {"status": True, "message": "", "data": records}
except pg.InternalError as e:
return {"status": False, "message": str(e), "data": None}
def run_upsert_query(conn, q, params):
try:
cur = conn.cursor()
cur.execute(q, tuple(params))
conn.commit()
id = cur.fetchone()[0]
return {"status": True, "message": "", "data": id}
except pg.InternalError as e:
conn.rollback()
return {"status": False, "message": str(e), "data": None}
def run_delete_query(conn, q, params):
try:
cur = conn.cursor()
cur.execute(q, tuple(params))
conn.commit()
return {"status": True, "message": "", "data": None}
except pg.InternalError as e:
conn.rollback()
return {"status": False, "message": str(e), "data": None}
Наконец, я бы предпочел предоставить свои ресурсы как формат JSON. Для этого я создал промежуточное программное обеспечение в корневом каталоге для Falcon.
import json
from datetime import date, datetime
class JSONEncoder(json.JSONEncoder):
def default(self, obj):
if isinstance(obj, bytes):
return obj.decode("utf-8")
if isinstance(obj, (date, datetime)):
return str(obj.isoformat())
return super(JSONEncoder, self).default(obj)
class ContentEncodingMiddleware(object):
def process_response(self, req, resp, _, req_succeeded):
if not req_succeeded:
return
if req.client_accepts_json:
resp.set_header('Content-Type', 'application/json')
resp.body = json.dumps(resp.body, cls=JSONEncoder)
Давайте создадим план по почтельону.
Вы сможете найти коды и связанные с этим репозиторий по этой ссылке.
https://github.com/mertingen/python-falcon-framework-api
Falcon Framework создает довольно крутые API. Надеюсь, эта статья будет полезна для вас и увидимся в следующей статье. Если у вас есть какие-либо проблемы, не стесняйтесь спрашивать.
Оригинал: “https://dev.to/_mertsimsek/falcon-api-framework-on-docker-2-e9k”