Создайте молниеносную систему напоминаний по SMS с Vonage SMS API и FastAPI Backend!

В этой статье вы узнаете, как создать серверную систему смс-напоминаний с помощью Vonage SMS API.

FastAPI — это современная, быстрая (высокопроизводительная) веб-инфраструктура для создания API с Python 3.6+ на основе стандартных подсказок типов Python. Он построен на основе Starlette для веб-частей и Pydantic для частей данных. Его преимущества заключаются в возможности быстрого создания конечных точек, встроенной поддержке веб-сокетов и асинхронности.

Я бы использовал Python для серверной системы из-за его простоты и масштабируемости, а также его обширной экосистемы и библиотек, которые можно использовать для широкого круга задач.

Отказ от ответственности: я рекомендую вам ознакомиться с официальной документацией FastAPI или видео tutorial, чтобы начать работу с фреймворком; так как это поможет ускорить процесс.

Vonage SMS API позволяет отправлять и получать текстовые сообщения пользователям по всему миру, используя наши REST API. Начните работу здесь.

Настройка учетной записи Vonage

Чтобы следовать этому руководству, я настоятельно рекомендую вам создать учетную запись Vonage. Вы можете начать с регистрации здесь и начать работу с бесплатным кредитом. После этого возьмите свой API и секретный ключ в верхней части панели инструментов Vonage API.

Настройка и установка проекта

1). Создайте и активируйте виртуальную среду

Для начала начните с создания каталога, виртуального окружения (желательно pipenv) и активируйте его:

mkdir sms_reminder && cd sms_reminder
python3.9 -m pipenv shell

2). Установите необходимые зависимости

Создайте файл requirements.txt в каталоге и продолжите, скопировав и вставив приведенный ниже фрагмент кода в файл requirements.txt:

aiosqlite==0.18.0
alembic==1.9.2
anyio==3.6.2
apscheduler==3.9.1.post1
backoff==2.2.1
certifi==2022.12.7
click==8.1.3
databases==0.7.0
fastapi==0.89.1
greenlet==2.0.1; python_version >= '3' and (platform_machine == 'aarch64' or (platform_machine == 'ppc64le' or (platform_machine == 'x86_64' or (platform_machine == 'amd64' or (platform_machine == 'AMD64' or (platform_machine == 'win32' or platform_machine == 'WIN32'))))))
h11==0.14.0
httpcore==0.16.3
httptools==0.5.0
httpx==0.23.3
idna==3.4
mako==1.2.4
markupsafe==2.1.2
pydantic==1.10.4
python-decouple==3.7
python-dotenv==0.21.1
pytz-deprecation-shim==0.1.0.post0
pytz==2022.7.1
pyyaml==6.0
rfc3986==1.5.0
six==1.16.0
sniffio==1.3.0
sqlalchemy==1.4.46
starlette==0.22.0
typing-extensions==4.4.0; python_version < '3.10'
tzdata==2022.7; python_version >= '3.6'
tzlocal==4.2
uvicorn==0.20.0
uvloop==0.17.0
watchfiles==0.18.1
websockets==10.4

Приступайте к немедленной установке необходимых пакетов с помощью следующей команды:

pipenv install -r requirements.txt

3). Создайте модуль main

Теперь, когда вы установили необходимые зависимости, давайте создадим файл main.py и вставим в него приведенный ниже код.

Это точка входа серверной системы, которая будет отвечать за построение приложения FastAPI, регистрацию маршрутизаторов API, событий для подключения и отключения базы данных, запуска планировщиков заданий и запуска приложения.

# Uvicorn Imports
import uvicorn

# FastAPI Imports
from fastapi import FastAPI


# construct application
app = FastAPI(
    title="SMS Reminder System",
    description="An SMS reminder system that allows users to text a specific number to set reminders for themselves.",
    verison=1.0,
)


@app.on_event("startup")
async def startup():
    pass # connect to database will come here


@app.on_event("shutdown")
async def disconnect():
    pass # disconnect from the database will come here


if __name__ == "__main__":
    uvicorn.run(
        "sms_reminder.main:app", host="0.0.0.0", port=3030, reload=True
    )

4). Создайте каталог models

Мы создали main.py, точку входа нашего приложения. Далее будет создан каталог моделей, и в основном там будут таблицы нашей базы данных, определенные как классы.

Давайте начнем, создадим файл __init__.py, чтобы пометить этот каталог как модуль Python.

«Чистый код пишется не по набору правил. Вы не станете мастером программного обеспечения, изучив список эвристик. Профессионализм и мастерство исходят из ценностей, определяющих дисциплину».

Создайте файл base.py, содержащий класс с именем ObjectTracker, который служит базовым классом для всех отслеживаемых объектов в базе данных.

# Stdlib Imports
from datetime import datetime

# SQLAlchemy Imports
from sqlalchemy import Column, Integer, DateTime


class ObjectTracker(object):
    id = Column(Integer, primary_key=True, index=True)
    date_created = Column(DateTime, default=datetime.now)
    date_updated = Column(DateTime, onupdate=datetime.now)

Чтобы создать базовый класс для отслеживания объектов в базе данных, мы импортируем классы Column, Integer и DateTime из библиотеки sqlalchemy, а также импортировать datetime из стандартной библиотеки Python, чтобы записывать время создания и обновления объектов.

Далее создадим файл sms.py, скопируем и вставим в него приведенный ниже код:

# SQLAlchemy Imports
from sqlalchemy import Column, String, DateTime

# Own Imports
from sms_reminder.models.base import ObjectTracker


class Reminder(ObjectTracker):
    phone_number = Column(String)
    message = Column(String)
    remind_when = Column(DateTime)

Класс Reminder представляет таблицу напоминаний в нашей базе данных, но мы должны настроить нашу базу данных, чтобы преобразовать этот объект в таблицу.

5). Создайте каталог config

Чтобы гарантировать, что Python рассматривает этот каталог как модуль, мы создаем файл __init__.py. Затем мы создаем файл database.py, который содержит информацию для настройки базы данных, построения создателя сеанса, коннектора базы данных (для подключения и закрытия нашей базы данных) и создания декларативного базового класса, который преобразует наши модели в таблицы базы данных.

Давайте продолжим; продолжите, скопировав приведенный ниже код и вставив его в файл database.py:

# SQLAlchemy Imports
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker, scoped_session
from sqlalchemy.ext.declarative import declarative_base

# Third Imports
from databases import Database


# create a database and engine
DATABASE_URL = "sqlite:///./sms_reminder.sqlite"
DATABASE_ENGINE = create_engine(
    DATABASE_URL, connect_args={"check_same_thread": False}
)

# construct a session maker
session_factory = sessionmaker(
    autocommit=False, autoflush=False, bind=DATABASE_ENGINE
)

# construct a scoped session
SessionLocal = scoped_session(session_factory)

# Construct a base class for declarative class definitions
Base = declarative_base()

# Construct a db connector to connect, and shutdown the database
db_connect = Database(DATABASE_URL)

Вы можете подумать: "Сейчас все может показаться довольно сложным". Но не волнуйтесь, я объясню магию, свидетелем которой вы являетесь.

* create_engine: этот метод принимает URL-адрес соединения (базы данных) и возвращает механизм sqlalchemy, который ссылается как на диалект, так и на пул, которые вместе интерпретируют функции модуля DBAPI, а также поведение базы данных ( Источник). * session_maker: чтобы объяснить это, нам нужно начать с понимания того, что такое Session. Сессия в sqlalchemy устанавливает все диалоги с базой данных, это обычный класс Python, который может быть создан напрямую. session_maker используется для создания конфигурации Session верхнего уровня, которую затем можно использовать во всем приложении без необходимости повторения определенных конфигурационных аргументов. Подробнее читайте здесь и здесь. * scoped_session: согласно официальной документации sqlalchemy, объект scoped_session создает управляемый потоком реестр объектов Session. Он обычно используется в веб-приложениях, так что одну глобальную переменную можно использовать для безопасного представления транзакционных сеансов с наборами объектов, локализованных в одном потоке. Короче говоря, они пытаются сказать, чтобы использовать scoped_session в вашем приложении для безопасности потоков. Поскольку он открывает и закрывает объект сеанса для каждого запроса, это безопасный и рекомендуемый способ работы. Подробнее читайте здесь. * declarative_base: эта функция используется для определения классов, сопоставленных с таблицами реляционной базы данных. * База данных: этот класс создает коннектор базы данных, который поможет нам подключиться и закрыть нашу базу данных.

Теперь, когда вы понимаете, что происходит, давайте добавим Base в класс Reminder в моделях sms:

# SQLAlchemy Imports
from sqlalchemy import Column, String, DateTime

# Own Imports
from sms_reminder.config.database import Base # new line
from sms_reminder.models.base import ObjectTracker


class Reminder(ObjectTracker, Base): # added Base
    __tablename__ = "reminders" # new line

    phone_number = Column(String)
    message = Column(String)
    remind_when = Column(DateTime)

Сделав это, sqlalchemy сопоставит определенный класс с таблицей реляционной базы данных. Затем мы обновляем точку входа main.py, чтобы подключаться и выключать нашу базу данных всякий раз, когда наш внутренний сервер генерирует событие запуска и завершения работы.

# Uvicorn Imports
import uvicorn

# FastAPI Imports
from fastapi import FastAPI

# Own Imports
from sms_reminder.config.database import db_connect # new line


# construct application
app = FastAPI(
    title="SMS Reminder System",
    description="An SMS reminder system that allows users to text a specific number to set reminders for themselves.",
    verison=1.0,
)


@app.on_event("startup")
async def startup():
    scheduler.start()
    await db_connect.connect() # new line


@app.on_event("shutdown")
async def disconnect():
    await db_connect.disconnect() # new line


if __name__ == "__main__":
    uvicorn.run(
        "sms_reminder.main:app", host="0.0.0.0", port=3030, reload=True
    )

Продолжайте, создав файл settings.py для хранения и хранения наших переменных окружения. Скопируйте и вставьте следующий код в файл:

# Stdlib Imports
from functools import lru_cache
from pydantic import BaseSettings

# Third Party Imports
from decouple import config as environ


class Settings(BaseSettings):
    """
    Settings to hold environmental variables.
    """

    VOYAGE_API_KEY: str = environ("VOYAGE_API_KEY", cast=str)
    VOYAGE_SECRET_KEY: str = environ("VOYAGE_SECRET_KEY", cast=str)


@lru_cache
def get_setting_values() -> Settings:
    env_var = Settings()
    return env_var

Давайте пробежимся по тому, что мы сделали:

• lru_cache: этот декоратор создаст объект Settings один раз. После этого он будет кэширован и использован повторно, вместо того чтобы создавать объект Settings каждый раз, когда мы хотим получить доступ к значениям.
• BaseSettings: позволяет переопределять значения переменными среды. Это полезно в производстве для секретов, которые вы не хотите сохранять в коде, это прекрасно работает с docker(-compose), Heroku и любым 12-факторным дизайном приложения.

Объект Settings отвечает за получение переменных среды, нам нужно создать файл .env, который будет загружен environ. Скопируйте и вставьте приведенный ниже код в файл:

VOYAGE_API_KEY=value
VOYAGE_SECRET_KEY=value

Замените value правильными ключами из вашей панели инструментов Vonage.

6). Инициализируйте alembic для обработки миграции базы данных

Alembic – это облегченный инструмент миграции базы данных, который можно использовать с набором инструментов SQLAlchemy Database Toolkit для Python. Он широко используется для миграции баз данных. Давайте приступим к его использованию.

Мы установили alembic в начале нашего проекта, поэтому теперь нам нужно инициализировать его для нашего рабочего каталога проекта. Запустите следующую команду в своем терминале:

alembic init migrations

Я использовал миграции для хранения всех миграций базы данных в этой папке. Выполнение приведенной выше команды создаст папку с именем миграции, которая содержит следующие файлы/папки:

- env.py
- README
- script.py.mako
- versions (directory)

В каталоге проекта также будет создан файл alembic.ini. В вашем каталоге versions не будет файлов версий, потому что мы еще не выполнили миграцию. Теперь, чтобы использовать перегонный куб, нам нужно внести некоторые изменения.

Сначала измените sqlalchemy.url в файле alembic.ini:

sqlalchemy.url = sqlite:///./sms_reminder.sqlite

Затем нам нужно предоставить нашу модель базы данных для alembic и получить доступ к метаданным из модели.

Отредактируйте файл env.py в папке migrations:

... # this means there are other imports here
# Own Imports
from sms_reminder.models.sms import Reminder # new line


... # this means there are other codes above
target_metadata = Reminder.metadata # update

Как показано выше, мы должны предоставить нашу модель базы данных для alembic. Запустите следующую команду в своем терминале:

alembic revision --autogenerate -m "Create reminder table"

Выполнение приведенной выше команды скажет alembic сгенерировать наш файл миграции и сохранить его в папке versions. После создания этого файла выполните следующую команду:

alembic upgrade head

Ваши таблицы будут сгенерированы в вашей базе данных. Это все к магии. Подробнее о перегонном кубе читайте здесь.

7). Создать каталог схем

Мы успешно создали классы models, представленные в виде таблиц базы данных, и выполнили миграцию базы данных. Далее будет создание наших пидантических моделей. Pydantic в основном используется для анализа и проверки данных с использованием аннотации типа.

Интересно отметить, что pydantic принудительно указывает тип во время выполнения и предоставляет удобные для пользователя ошибки, когда данные недействительны. Для начала создайте файл __init__.py, чтобы указать Python для обработки каталога как модуля, создайте другой файл с именем crud.py и вставьте в него следующие коды:

# Stdlib Imports
import pytz
from datetime import datetime
from pydantic import BaseModel, Field


class BaseReminderSchema(BaseModel):
    phone_number: str = Field(
        description="What's your phone number? Ensure you include your country code and is valid. E.g 234xxxxxxxxxx"
    )
    message: str = Field(
        description="What message do you want to remind yourself with? E.g Time to go to the gym!"
    )
    remind_when: datetime = Field(
        description="When should I send this message to you?",
        default=datetime.now(tz=pytz.timezone("Africa/Lagos")),
    )


class CreateReminderSchema(BaseReminderSchema):
    pass


class ReminderSchema(BaseReminderSchema):
    id: int
    date_created: datetime

    class Config:
        orm_mode = True

Давайте рассмотрим, что происходит в приведенном выше коде. Мы импортируем библиотеку часовых поясов python (pytz), дату и время из библиотеки datetime и импортируем BaseModel и Field из pydantic.

• pytz: эта библиотека позволяет выполнять точные межплатформенные расчеты часовых поясов с использованием Python 2.4 или более поздней версии. Это также решает проблему неоднозначного времени в конце летнего времени. Подробнее читайте здесь.
• BaseModel: это класс, используемый для определения объектов в pydantic. Вы можете думать о моделях как о типах в строго типизированных языках или как о требованиях одной конечной точки в API. Подробнее читайте здесь.
• Поле: это класс, используемый для предоставления дополнительной информации о поле либо для схемы модели, либо для комплексной проверки.
• BaseReminderSchema: это класс, наследуемый от BaseModel для создания полей, которые будут использоваться в других схемах.
• CreateReminderSchema: это класс, наследуемый от BaseReminderSchema, который будет использоваться в качестве требования для создания напоминания.
• ReminderSchema: это класс, наследуемый от BaseReminderSchema, который будет использоваться в качестве требования для составления списка созданных напоминаний. orm_mode (также известный как экземпляры произвольного класса) поддерживает модели, которые сопоставляются с объектами ORM.

8). Создать каталог interface

Нам нужен интерфейс, который находится между нашим уровнем доступа к данным (базой данных) и уровнем бизнес-логики (сервисами) приложения. Лучший способ сделать это — использовать шаблон проектирования репозитория. Видите ли, шаблон репозитория абстрагирует способ, которым ваши данные запрашиваются или создаются для вас.

Двигаясь дальше, создайте файл __init__.py, чтобы указать Python рассматривать каталог как модуль, а затем создайте файл с именем sms.py. Скопируйте и вставьте приведенный ниже код в файл:

# Stdlib Imports
from typing import List
from datetime import datetime

# SQLAlchemy Imports
from sqlalchemy.orm import Session

# Own Imports
from sms_reminder.models.sms import Reminder
from sms_reminder.config.database import SessionLocal


class ReminderORMInterface:
    """Reminder ORM that interface with the database."""

    def __init__(self) -> None:
        self.db: Session = self.get_db_session().__next__()

    def get_db_session(self):
        """
        This method creates a database session, 
        yields it to the caller, rollback the session 
        if an exception occurs; otherwise, close the session.
        """
        session = SessionLocal()
        try:
            yield session
        except Exception:
            session.rollback()
        finally:
            session.close()

    async def get(self) -> List[Reminder]:
        """This method gets a list of reminders."""
        reminders = self.db.query(Reminder).all()
        return reminders

    async def create(
        self, phone_number: str, message: str, remind_when: datetime
    ) -> Reminder:
        """This method creates a new reminder to the database."""
        reminder = Reminder(
            phone_number=phone_number,
            message=message,
            remind_when=remind_when,
        )
        self.db.add(reminder)
        self.db.commit()
        self.db.refresh(reminder)
        return reminder


reminder_orm = ReminderORMInterface()

Давайте разберем, что происходит в приведенном выше коде, у нас есть четыре метода:

• get_db_session: этот метод создает для нас сеанс базы данных, возвращает его и откатывает сеанс в случае возникновения исключения; в противном случае закройте сеанс.
• __init__: этот метод инициализирует сеанс базы данных.
• get: этот метод запрашивает модель Reminder (таблицу), получая список всех объектов в ней.
• create: этот метод создает новый экземпляр напоминания, сохраняет его в состоянии (которое будет либо откатываться, либо двигаться вперед), сбрасывать ожидающие изменения и фиксировать текущую транзакцию, а также обновлять вновь созданный экземпляр в стол.

Чтобы начать использовать наш интерфейс, нам нужно инициализировать наш класс переменной.

9). Создайте каталог services

В предыдущем разделе мы рассмотрели, что такое интерфейс (шаблон репозитория); и как он расположен между нашим уровнем доступа к данным и уровнем бизнес-логики нашего приложения. В этом разделе мы создадим уровень бизнес-логики. Двигаясь дальше, создайте файл __init__.py, чтобы указать Python рассматривать каталог как модуль.

я). Создайте файл с именем vonage.py, скопируйте и вставьте в него приведенный ниже код:

# Stdlib Imports
from typing import Dict

# FastAPI Imports
from fastapi import HTTPException

# Own Imports
from sms_reminder.config.settings import get_setting_values as settings

# Third Party Imports
import httpx
import backoff


class VonageSMS:
    """SMS API Service provider to handle sending text messages."""

    def __init__(self) -> None:
        """
        The method is used to initialize the class
        and set the base_url, settings, secret_key and api_key.
        """
        self.base_url = "https://rest.nexmo.com/sms/json"
        self.settings = settings()
        self.secret_key = self.settings.VOYAGE_SECRET_KEY
        self.api_key = self.settings.VOYAGE_API_KEY

    def headers(self) -> Dict[str, str]:
        """
        This method returns a header with a key of "Content-Type" and a value of
        "application/x-www-form-urlencoded".
        """
        return {"Content-Type": "application/x-www-form-urlencoded"}

    @backoff.on_exception(backoff.expo, httpx.ConnectTimeout, max_time=100)
    async def send(self, phone_number: str, message: str) -> True:
        """
        This method sends a message to the provided phone number using the VonageSMS API.
        :param phone_number: The phone number to send the message to
        :type phone_number: str
        :param message: The message you want to send
        :type message: str
        :return: True
        """
        async with httpx.AsyncClient() as client:
            request_data = f"from=Send Reminder!&text={message}&to={phone_number}&api_key={self.api_key}&api_secret={self.secret_key}"
            response = await client.post(
                url=self.base_url, headers=self.headers(), data=request_data
            )
            response_data = response.json()["messages"][0]
            if response_data["status"] == "0":
                return True
            raise HTTPException(
                500,
                {
                    "source": "VonageSMS",
                    "message": response_data["error-text"],
                },
            )


vonage_sms = VonageSMS()

"Ух ты. Абрам, много кода!" Не беспокойтесь об этом, давайте рассмотрим это вместе. Один импорт, метод за раз. Я не буду объяснять общие вещи, которые, как я полагаю, вы уже должны знать, такие как Dict из typing.

* HTTPException: это базовый класс для всех исключений http. Вы можете переопределить его для создания пользовательских исключений, таких как ServerException, NotFoundException и т. д. * settings: это функция из модуля config/settings.py, которая при вызове кэширует наши переменные среды, а также предоставляет нам доступ к значению. * httpx: это полнофункциональная клиентская http-библиотека для python3. Он поддерживает как http/1.1, так и http/2 и предоставляет как синхронизирующие, так и асинхронные API. * backoff: эта библиотека предоставляет декораторы функций, которые можно использовать для обертывания функции таким образом, что она будет повторяться до тех пор, пока не будет выполнено какое-либо условие.

Вы видите это сейчас? Не нужно было паниковать, я здесь, чтобы все правильно объяснить. Далее, что делает каждый метод:

* __init__: этот метод используется для инициализации нашего класса VonageSMS и установки base_url и настроек; secret_key и api_key смс-сервиса vonage. * headers: этот метод используется для установки типа контента, который мы хотим в наших заголовках, когда мы делаем запрос к службе смс Vonage через API. * send: этот метод отправляет сообщение на номер телефона, указанный с помощью Vonage sms api. Чтобы иметь возможность отправить смс, нам нужен номер телефона и сообщение, которое мы хотим отправить.

Обратите внимание на использование httpx.AsyncClient() вместо httpx.Client в методе. Это связано с тем, что AsyncClient является асинхронным, а его диспетчер контекста использует асинхронный с, а не только с.

* Двигаясь вперед, мы кодируем форму в URL-адрес и отправляем почтовый запрос на наш base_url, который мы установили в нашем методе __init__. Затем мы проверяем, есть ли у нас статус, равный "0", возвращаем True, иначе вызываем исключение 500, сообщающее клиенту, что пошло не так.< /p>

ii). Создайте файл с именем tasks.py, скопируйте и вставьте в него приведенный ниже код:

# Stdlib Imports
from random import randint
from datetime import datetime

# APScheduler Imports
from apscheduler.schedulers.asyncio import AsyncIOScheduler

# Own Imports
from sms_reminder.services.vonage import vonage_sms


# initialize ayncio scheduler
scheduler = AsyncIOScheduler()


async def create_reminder_job(
    phone_number: str, message: str, remind_when: datetime
) -> dict:
    """
    This function creates a job that will send a message to the provided 
    phone number at a specific time.

    :param phone_number: The phone number to send the message to
    :type phone_number: str

    :param message: The message to be sent
    :type message: str

    :param remind_when: The datetime object that represents the time when the reminder should be sent
    :type remind_when: datetime


    :return: A dictionary with the keys "scheduled" and "job_id".
    """

    job_uid = randint(0, 9999)
    reminder_job = scheduler.add_job(
        func=vonage_sms.send,
        trigger="date",
        args=(phone_number, message),
        name=f"reminder_set_{phone_number}_{job_uid}",
        id=f"{phone_number}_{job_uid}",
        next_run_time=remind_when,
    )
    return {"scheduled": True, "job_id": reminder_job.id}

Мы импортируем AsyncIOScheduler из библиотеки apscheduler и импортируем vonage_sms из нашего модуля services/vonage. Функция create_reminder_job отвечает за создание задания, которое отправит сообщение на указанный номер телефона по определенному номеру.

III). Создайте файл с именем sms.py, скопируйте и вставьте в него следующий код:

# Stdlib Imports
from datetime import datetime
from typing import List

# FastAPI Imports
from fastapi import HTTPException

# Own Imports
from sms_reminder.services.tasks import create_reminder_job
from sms_reminder.interface.sms import reminder_orm, Reminder


async def create_user_reminder(
    phone_number: str, message: str, remind_when: datetime
) -> Reminder:
    """
    This function creates a new reminder in the database.

    :param phone_number: The phone number of the user who will receive the reminder
    :type phone_number: str

    :param message: The message that will be sent to the user
    :type message: str

    :param reminder_when: datetime
    :type reminder_when: datetime

    :return: A reminder object
    """

    reminder = await reminder_orm.create(phone_number, message, remind_when)
    reminder_job = await create_reminder_job(
        phone_number, message, remind_when
    )
    if reminder_job["scheduled"]:
        return reminder
    raise HTTPException(400, {"message": "Was not able to set reminder!"})


async def get_user_reminders() -> List[Reminder]:
    """
    This function gets the list of user reminders.

    :return: A list of reminder objects.
    """

    reminders = await reminder_orm.get()
    return reminders

Мы импортируем create_reminder_job из модуля services/tasks, класса модели reminder_orm и Reminder из interface/sms модуль. Функция create_user_reminder отвечает за создание нового напоминания в базе данных. Когда мы создаем новое напоминание, мы также создаем задание, которое отправит сообщение с напоминанием на указанный номер телефона. Исключение 400 возникает, если мы не смогли установить задание.

Далее следует функция get_user_reminders, по сути, мы получаем список напоминаний из базы данных.

10). Создайте каталог api

В предыдущем разделе мы создали уровень бизнес-логики. Следующим шагом будет создание конечных точек нашего API. Для начала создайте файл __init__.py, чтобы указать Python рассматривать каталог как модуль.

я). Создайте файл с именем index.py, скопируйте и вставьте в него приведенный ниже код:

# FastAPI Imports
from fastapi import APIRouter


# initialize api router
router = APIRouter(tags=["Root"])


@router.get("/")
async def root() -> dict:
    """
    SMS Reminder System root

    Returns: dict: version and description
    """

    return {
        "version": 1.0,
        "description": "An SMS reminder system that allows users to text a specific number to set reminders for themselves.",
    }

APIRouter — это класс, импортированный из fastapi, мы используем его для объявления наших операций с путями. root — это представление API, которое возвращает версию и описание нашего серверного приложения.

ii). Создайте файл с именем crud.py, скопируйте и вставьте в него приведенный ниже код:

# Stdlib Imports
from typing import List
from datetime import datetime, timezone

# FastAPI Imports
from fastapi import APIRouter, status, HTTPException

# Own Imports
from sms_reminder.services.sms import create_user_reminder, get_user_reminders
from sms_reminder.schemas.crud import CreateReminderSchema, ReminderSchema


# initialize the api router
router = APIRouter(tags=["Reminder"], prefix="/reminders")


@router.post("/", status_code=status.HTTP_201_CREATED)
async def create_reminder(payload: CreateReminderSchema):
    """
    This API view creates and sets a reminder.
    :param payload: CreateReminderScheman
    :type payload: CreateReminderSchema

    :return: The reminder is serialized to a JSON.
    """

    if payload.remind_when <= datetime.now(timezone.utc):
        raise HTTPException(
            400,
            {
                "message": "Kindly set a date and time that exceeds the past and now."
            },
        )

    reminder = await create_user_reminder(
        payload.phone_number, payload.message, payload.remind_when
    )
    return {"message": "Reminder set!", "data": reminder}


@router.get("/", response_model=List[ReminderSchema])
async def get_reminders():
    """
    This API view gets the total available reminders.

    :return: A list of reminders.
    """

    reminders = await get_user_reminders()
    return reminders

Мы рассмотрели, что такое APIRouter и HTTPException; двигаясь дальше, мы объявили HTTP-код status, который мы хотели бы получить в нашем ответе.

Мы также импортируем нашу бизнес-логику, функции create_user_remeinder и get_user_reminders из модуля services/sms и, наконец; импорт CreateReminderSchema и ReminderSchema из модуля schemas/crud.

Представление API create_reminder отвечает за создание и настройку напоминания. Чтобы объявить метод HTTP, который мы хотим использовать, мы используем маршрутизатор в качестве декоратора для доступа к методу post.

Ключевое слово async в определении функции сообщает FastAPI, что она должна выполняться асинхронно, то есть без блокировки текущего потока выполнения. Мы используем CreateReminderSchema в качестве аннотации типа для параметра payload.

Оператор if является проверкой, чтобы убедиться, что мы не можем создать напоминание с прошедшей датой и временем. Двигаясь вперед, create_user_reminder вызывается для управления созданием и настройкой напоминания. Если исключение не возникло, ответом API будет словарь с сообщением и напоминанием.

12). Запустить сервер Запустите сервер uvicorn с помощью следующей команды в вашем терминале:

python sms_reminder/main.py

ПРИМЕЧАНИЕ. Если вы получаете следующую ошибку:

Traceback (most recent call last): 
File "/.../sms-based-reminder-system/sms_reminder/main.py", line 8, in from sms_reminder.services.tasks import scheduler 
ModuleNotFoundError: No module named 'sms_reminder'

Запустите приведенный ниже код в своем терминале и снова запустите сервер uvicorn:

export PYTHONPATH=$PWD

В консоли терминала вы получите что-то похожее на приведенное ниже сообщение:

INFO: Will watch for changes in these directories: [‘/…/sms-based-reminder-system’]
INFO: Uvicorn running on  [http://0.0.0.0:3030](http://0.0.0.0:3030/)  (Press CTRL+C to quit)
INFO: Started reloader process [1203267] using WatchFiles
INFO: Started server process [1203280]
INFO: Waiting for application startup.
INFO: Application startup complete.

Заключение

Поздравляем! Вы узнали, как создать серверную систему SMS-напоминаний с использованием FastAPI и Vonage SMS API. Есть много других вещей, которые вы можете сделать, например, ограничить API создания напоминаний для обработки 10 запросов в минуту, чтобы уменьшить количество спам-ботов. Вы можете найти исходный код этого учебного проекта на моем Github.