Обложка фото Крис Рид на Неспособный
FASTAPI это рамка Python для строительства RESTFUL API. Он имеет всю простоту Python с дополнительными преимуществами Async⚡, генерации автоматической схемы и типов Openapi и Python (с Pydantic). Fastapi – относительно новый проект и набирает довольно хорошую тягу в мире разработчиков. У него почти 16K звезд на GitHub и около 160 участников. Основная плюс сторона Fastapi – это документация, которая является Верхняя выемка Анкет
Deta является компанией, которая создает облачную платформу, которая является дружелюбной для разработчиков и уменьшит хлопот от идеи к приложению.
” Deta строит облако для разработчиков с меньшим количеством «наворотов и свистков», только то, что нужно, чтобы выполнить работу – микро облако ».
Лично то, что привлекло меня к Дету, было простой для реализации. Мне было трудно соединить Fastapi с SQLalchemy, так как я нуб. Затем Детя вошел или, по крайней мере, тогда я узнал об этом.
Сначала я был скептически настроен, но простота работы с Deta поразила меня, и их общество сообщество – ✨. Создатели почти активны каждый раз, и они помогут вам в любом случае, чтобы преодолеть проблему.
Что мы построим
Мы собираемся построить личный Список рассылки API с
- FASTAPI для API и логики
- Deta.sh как наш DB
- Развернуть его на Deta Micros
Мы только собираемся построить API, вы можете использовать Mailgun или SendGrid отправить электронные письма. Наш API будет состоять из 4 операций.
- Добавьте нового человека в наш список
- Получите все электронные письма из нашего списка
- Обновите электронное письмо конкретного пользователя в нашем списке
- Удалить подписчика/пользователя из нашего списка
Настройка FASTAPI и DETA
Надеюсь, у вас есть интерпретатор Python 3.6+ на вашей машине. Если вы не обязательно загрузите его, установите и проверьте, запустив Python в вашем терминале
$ python3 Python 3.7.6 (default, Dec 30 2019, 19:38:26) [Clang 11.0.0 (clang-1100.0.33.16)] on darwin Type "help", "copyright", "credits" or "license" for more information. >>>
Вышеуказанный вывод находится на моей машине, и он будет переходить от машины на машину.
Следующим является иметь папку и виртуальную среду, чтобы мы могли бы использовать либо Pipenv, либо Venv. Pipenv является пакетом для автоматического управления и управления Python Envs, вы можете установить Pipenv с,
$ pip install --user pipenv
Как только мы установили Pipenv, чтобы настроить нашу среду, следуйте командам для установки
- FASTAPI
- Uvicorn – сервер ASGI
- Deta – обертка Python для баз Deta
$ mkdir mailing-list $ cd mailing-list $ pipenv shell #it will make the virtual environment $ pipenv install fastapi uvicorn deta
Если мы используем Venv, то для создания Venv и установки пакетов следуйте
$ mkdir mailing-list $ cd mailing-list $ python3 -m venv venv $ source venv/bin/activate $ pip install fastapi uvicorn deta
Они предназначены для систем Unix и в Windows, чтобы активировать виртуально –
venv\Scripts\activate
Как только мы настроили нашу среду разработки. Давайте перейдем к созданию списка рассылки.
Для DETA создайте учетную запись в DETA, создайте новый проект и получите ключ проекта. Сохраните ключ проекта, так как он будет показан один раз. Вы можете сохранить его как переменную среды, выполнив приведенный ниже код в Unix Like (Linux/Mac).
export DETA_PROJECT_KEY =
и в окнах, используйте
set DETA_PROJECT_KEY =
Мы также сохраним ключ в .env
файл. Этот файл env важен для процесса развертывания. Для этого нам нужно создать файл с именем .env
и сохранить ключ проекта, как
DETA_PROJECT_KEY =
Вот и все. Давайте перейдем к коду
Код
Давайте начнем с того, что сделаем простой конечную точку API REST с FastAPI в файле с именем main.py
# main.py from fastapi import FastAPI app = FastAPI() @app.get("/") async def helloworld(): return "Hello World"
Это всего лишь приложение Hello World с Fastapi, и для того, чтобы запустить это, нам понадобится Uvicorn. Выполнить следующее в терминале/CMD,
$ uvicorn main:app --reload INFO: Uvicorn running on http://127.0.0.1:8000 (Press CTRL+C to quit) INFO: Started reloader process [13050] using statreload INFO: Started server process [13052] INFO: Waiting for application startup. INFO: Application startup complete.
Наш API будет действовать в порт 8000
Анкет Fastapi обладает встроенной поддержкой Openapi, Swagger и Redoc. Чтобы просмотреть документы для нашего приложения, просто перейдите к Localhost: 8000/документы . Чтобы проверить, работает ли наша конечная точка, Просто выполнить
$ curl http://127.0.0.1:8000/ "Hello World"
Итак, давайте перейдем к нашим основным частям
Новый подписчик
Мы будем использовать Документы от DETA для всех ссылок.
Нам нужно создать новую конечную точку, чтобы кто -то подпишился на наш список рассылки. Так что нам просто понадобится имя человека и электронная почта, чтобы быть в нашем БД.
Позаботьтесь о том, чтобы в некоторых странах незаконно отправлять электронные письма людям без согласия. Мы сделаем удаленную конечную точку, чтобы они удалили себя из списка
Давайте перейдем к коду для добавления нового подписчика в наш список. Код Python без использования DB или просто кода с FASTAPI.
# main.py from fastapi import FastAPI from pydantic import BaseModel, NameEmail app = FastAPI() class Subscriber(BaseModel): full_name: str email: NameEmail @app.post("/subscribe") async def newUser(sub: Subscriber): name = sub.full_name email = sub.email # you could also return sub.json() or sub.dict() return { "full_name": name, "email": email }
Мы используем Pydantic Базодель
класс в наследуя Наш Подписчик
класс. Затем мы делаем дикт абонента, чтобы получить данные из тела запроса. Nameemail
Поможет нам получить имя часть электронной почты.
Вывод – это объект name -mail, который имеет два свойства: имя и электронная почта. Для fred.bloggs@example.com именем недвижимость будет “fred.bloggs”.
Поскольку мы используем Nameemail Нам нужно установить Электронный валидатор
работать с этим. Просто беги PIP установить Pydantic [Email-Validator]
Этот фрагмент вернет вещи, которые мы предоставили ему. Я ввел данные, перейдя в Localhost: 8000/документы Если у вас работает сервер. Наш вывод, если мы предоставм Athul
для имени и athul@example.com
по электронной почте будет
{ "name": "athul", "email": "athul@example.com", }
Теперь давайте подключим его с базой данных, базой Deta в этом случае. Мы будем использовать Python’s ОС
Пакет, чтобы получить ключ проекта. Мы сохранили ключ проекта на Env Vars и перезапустите сервер, чтобы он работал.
# main.py import os from fastapi import FastAPI from pydantic import BaseModel, NameEmail from deta import Deta deta = Deta(os.getenv("DETA_PROJECT_KEY")) db = deta.Base("mydb") # this can be anything related to your use case app = FastAPI() class Subscriber(BaseModel): full_name: str email: NameEmail @app.post("/subscribe") async def newUser(sub:Subscriber): subscriber = db.put({ "full_name": sub.full_name, "email": sub.email.email, "key": sub.email.name }) return subscriber
Здесь мы импортировали библиотеку DETA и инициализировали ее с именем DB. Мы указываем ключ для данных. Этот ключ будет частью электронной почты, поскольку в большинстве случаев он будет уникальным. Если мы не указам ключ, DETA генерирует уникальный ключ для записи.
Вы можете иметь несколько DBS с одним ключом проекта
Мы используем Deta’s Положить
Метод (/функция) для вставки ваших данных в нашу базу данных. Метод PUT будет генерировать уникальный ключ Для наших данных и будут храниться, если мы не указаем ключ. Вывод, если мы добавим себе нового подписчика с предоставлением Джон Доу
для имени и John@example.com
по электронной почте будет,
{ "email": "john@example.com", "key": "john", "full_name": "John Doe" }
Таким образом, мы сделали конечную точку для людей, чтобы подписаться на наш список рассылки 🎉. Далее мы сделаем конечную точку, чтобы люди могли обновить информацию о своей как их имя или электронная почта.
Обновите подписчик
Мы создадим новый Updatesubscriber
класс для обновления имени или электронной почты подписчика. Мы будем использовать получить
и Обновление
Методы DETA для получения и обновления данных. Мы также выразим ошибку, если запись не найдена в базе
# main.py from typing import Optional from fastapi import NameEmail,HTTPException class UpdateSub(BaseModel): full_name: Optional[str] = None email: Optional[NameEmail] = None @app.patch("/subscriber/{email}") async def update_sub(upsub: UpdateSub, email: NameEmail): sub = db.get(email.name) if sub is None: raise HTTPException(404,"User Not Subscribed") updates = {} if upsub.full_name: updates["full_name"] = upsub.full_name if upsub.email: updates["email"] = upsub.email.email db.update(updates, key=sub["key"]) return {"msg": "User Updated"}
Эта конечная точка обновит электронную почту или полную_name подписчика, но предостережение здесь заключается в том, что ключ не может быть изменен, даже если электронное письмо изменено. Давайте сделаем нового пользователя с именем Джона Риза (заинтересованного лица) с первоначальным электронным письмом reese@gmail.com
. Теперь давайте обновим его электронную почту с помощью этой конечной точки, для этого наш URL -адрес запроса будет следующим,
http://localhost:8080/subscriber/reese%40example.com
Вы можете увидеть это после /подписчик
В URL мы передаем исходное электронное письмо, и это называется Параметр пути Анкет Мы используем параметр Path и получаем его часть и используем его в качестве ключа. Для обновления электронного письма мы берем новое электронное письмо и обновляем его в нашей базе Deta с помощью команды Update. Обновления должны быть в форме питона диктат
. Если выполненная операция верна, тогда мы получим Нет
введите в возврат
Получите всех подписчиков
Эта конечная точка отправит Получить
Метод HTTP для извлечения всех данных из нашей базы DETA. Мы будем использовать Fetch
Функция/Метод данных для извлечения всех данных с нашей базы. Fetch
Функция вернет генератор Тип, поэтому мы будем использовать Next ()
Функция, чтобы получить все данные как Список
Анкет Конечная точка будет похожа на
@app.get("/subscribers") async def get_all_subs(): return {"subscribers": next(db.fetch())}
Вывод вышеуказанного запроса будет
{ "subscribers": [ { "email": "athul@example.com", "key": "athul", "name": "Athuk" }, { "email": "john@example.com", "full_name": "John Doe", "key": "john" }, { "email": "jhreese@gmail.com", "full_name": "John Reese", "key": "reese" } ] }
Теперь мы настроили конечные точки создания, обновление и чтение. Далее мы рассмотрим операцию удаления.
Удалить подписчики
Чтобы удалить подписчика, мы возьмем в full_name
как вход, чтобы найти подписчика И если у нас есть более одной записи с той же самой Full_Name, мы возьмем данные по электронной почте. Тогда мы будем использовать Удалить
Метод удаления записи, получая ключ из наших записей.
# main.py class DelSub(BaseModel): email: Optional[NameEmail] full_name: str @app.delete("/subscribers.del") async def unsubscribe(unsub: DelSub): data = delsub.dict() name = data.get('full_name') email = data.get('email') sub = next(db.fetch({'full_name':name})) if len(sub) > 1 or len(sub) == 0: sub = next(db.fetch({'email':email.email})) if len(sub)==0: raise HTTPException(400,"email not found") db.delete(sub[0]['key']) return {"msg":"Record Deleted"}
Ответ, если запись удалена, будет Python Нет
введите, поэтому мы отвечаем сообщением,
{"msg":"Record Deleted"}
Итак, мы сделали CRUD API для личного списка рассылки. Теперь давайте развернуть его.
Развернуть наш API
Для развертывания нашего API нам понадобится Deta Micros.
Delta Microservices) – это легкое, но масштабируемое время выполнения облака, привязанное к конечной точке HTTP. Они призваны поднять ваши приложения и бежать ярко быстро. Сосредоточьтесь на написании вашего кода, и Deta позаботится обо всем остальном.
Вы должны установить CLI Deta Micros ‘в вашей системе. Вы можете установить, следуя инструкциям в этом Ссылка Анкет После установки нам нужно войти в DETA через CLI, для выполнения следующей команды
deta login
Он откроет новое окно браузера и будет аутентифицировать CLI. Далее нам нужно начать новый микро. Нам также понадобится Требования.txt
Для управления нашими зависимостями, для этого вы можете скопировать ответ для команды PIP FREEZE
и вставьте его в файл с именем Требования.txt
или на *nix системы, Вы можете просто выполнить
pip freeze > requirements.txt
Это вставлено вывод PIP замораживает
к Требования.txt
Анкет Как только мы это сделаем, нам нужно развернуть его в Deta, используя команду. После того, как мы это сделаем Deta new командование
$ deta new { "name": "mailing-list", "runtime": "python3.7", "endpoint": "", "visor": "enabled", "http_auth": "enabled" }
Он автоматически обнаружит, что нам понадобится время выполнения Python, и будет развернуть наш код и даст нам уникальную ссылку для нашего Micro.
Yay, мы развернули API, для деталей Micro просто выполнить следующее для получения подробной информации о микро.
$ deta details { "name": "mailing-list", "runtime": "python3.7", "endpoint": "", "visor": "enabled", "http_auth": "enabled" }
Нам нужно обновить переменную среды, чтобы установить наш Project_key, выполнить Обновление Deta
командовать подобным
$ deta update -e .env Updating environment variables... Successfully updated micro's environment variables
Вы можете отключить аутентификацию, чтобы API был доступен, просто выполните Deta Auth Disable
Анкет
$ deta auth disable Successfully disabled http auth
Если мы перейдем к URL нашего Micro, мы увидим Привет, мир
Ответ, который мы написали как /
конечная точка. Мы можем увидеть спецификацию OpenAPI и документацию Swagger, если мы перейдем к /документы
URL, мы увидим такую страницу
Yay, мы создали API с Fastapi и Deta и развернули его в Интернете.
Это приводит к концу этого поста, вот весь исходный код
import os from typing import Optional from fastapi import FastAPI, HTTPException from pydantic import BaseModel, NameEmail from deta import Deta deta = Deta(os.getenv("DETA_PROJECT_KEY")) db = deta.Base('mailing-list') app = FastAPI() class Subscriber(BaseModel): full_name: str email: NameEmail @app.post("/subscribe") async def newUser(sub:Subscriber): data = sub.dict() name = data.get("full_name") email = data.get("email") subscriber= db.put({ "full_name":name, "email":email.email },key=email.name) return subscriber @app.get("/subscribers") async def getAllSubs(): return {"subscribers":next(db.fetch())} class UpdateSub(BaseModel): full_name: Optional[str] = None email: Optional[NameEmail] = None @app.patch("/subscriber/{email}") async def updateSub(subscriber:UpdateSub,email:NameEmail): data = subscriber.dict() new_name = data.get("full_name") new_email = data.get("email") sub = db.get(email.name) if sub is None: raise HTTPException(400,"User Not Subscribed") if new_email is None: update = {"full_name":new_name} if new_name is None: update = {"email":new_email.email} db.update(updates=update,key=sub['key']) return {"msg":"User Updated"} class DelSub(BaseModel): email:Optional[NameEmail] full_name:str @app.delete("/subscribers.del") async def deleteSub(delsub:DelSub): data=delsub.dict() name = data.get('full_name') email = data.get('email') sub = next(db.fetch({'full_name':name})) if len(sub) > 1 or len(sub) == 0: sub = next(db.fetch({'email':email.email})) if len(sub)==0: raise HTTPException(400,"email not found") db.delete(sub[0]['key']) return {"msg":"Record Deleted"} @app.get("/") async def helloworld(): return "Hello World"
Вывод
Это всего лишь демонстрация для начала работы с Fastapi с Deta.Sh. Вы можете изменить ключ для ваших предпочтений или использовать ключ, сгенерированный автоматически DETA. Вы также можете использовать адрес электронной почты в качестве ключа, но вам может потребоваться переключить @
персонаж с чем -то еще, как -
или _
за его место. Если у вас есть какие -либо вопросы, сомнения или отзывы, вы можете использовать раздел комментариев для того же самого и позвольте мне, если это помогло вам. Иди и постройте потрясающие вещи ⚡ ️. Спасибо ❤
Оригинал: “https://dev.to/athulcajay/fastapi-deta-ni5”