предисловие
Flask — это легкий веб-фреймворк, разработанный на Python. Насколько он легкий? Веб-сервис можно разработать менее чем в 10 строк, но это можно использовать только для демонстрации.Сегодня я потрачу 1 час на разработку микросервиса SMS для производственной среды. Ниже приведен служебный код, который доступен непосредственно после десенсибилизации нашей производственной среды, а не пример учебника.
Зачем разрабатывать микросервисы SMS?
Мы все полагаемся на реализацию общедоступного облака для сервисов SMS и вызываем их напрямую через API общедоступного облака, так зачем же нам самим его инкапсулировать?
- Поскольку мы хотим уменьшить количество дублирующихся кодов в среде микросервисов, если есть несколько микросервисов, которым необходимо использовать службу SMS, нам нужно копировать код несколько раз, а также обернуть API общедоступного облака в наш собственный API микросервиса для копирования. код, сокращенный до одной строки Http-запроса.
- Ключ доступа и секрет для вызова API не нужно копировать в несколько служб, что снижает риски безопасности.
- Общая бизнес-логика может быть добавлена в соответствии с потребностями нашего бизнеса.
Влияет ли еще один уровень вызовов на производительность?
Еще один уровень вызовов — это еще один сетевой запрос, но влияние минимально. Мы не можем писать построчный код, потому что слишком много вызовов объектно-ориентированным способом.
- Служба SMS общедоступного облака — это асинхронный вызов, и обработка ошибок также является методом асинхронного обратного вызова.
- Вызов внутренней сети микросервиса должен быть очень быстрым, и его можно развернуть на той же виртуальной машине или в том же машинном зале.
Начинать
Сначала мы строим скелет проекта.
Зачем создавать скелет проекта?
Поскольку Flask слишком легкий, такие спецификации, как конфигурация, маршрутизация и т. д., должны определяться самими разработчиками. Как правило, зрелая команда разработчиков имеет собственный набор скелетов разработки, которые должны быть настроены унифицированным образом, разработаны унифицированным образом и интегрированы со связанными системами. Здесь я делюсь очень простым скелетом разработки, подходящим для производственных сред.
Создайте новый каталог проекта, а затем создайте в нем два каталога Python app и config. app используется для хранения кода, связанного с бизнесом, а config используется для хранения кода, связанного с конфигурацией.
класс конфигурации
существуетconfig/config.pyДобавьте в конфиг следующее содержимое.Дизайн конфигурации варьируется от человека к человеку, и Flask не накладывает никаких ограничений. Мой дизайн здесь заключается в использовании BaseConfig в качестве базового класса конфигурации для хранения всех общих конфигураций, в то время как в разных средах используются разные подклассы конфигурации, и подклассам нужно только изменить определенные значения для удобства просмотра.
Если настроенное значение необходимо ввести во время выполнения (например, подключение к базе данных и т. д.), вы можете использовать способ переменных среды (например, SECRET_KEY ниже), я также используюorПредоставляются значения по умолчанию без переменных среды.
import os
class BaseConfig:
"""
配置基类，用于存放共用的配置
"""
SECRET_KEY = os.environ.get('SECRET_KEY') or os.urandom(16)
DEBUG = False
TESTING = False
class ProductionConfig(BaseConfig):
"""
生产环境配置类，用于存放生产环境的配置
"""
pass
class DevelopmentConfig(BaseConfig):
"""
开发环境配置类，用于存放开发环境的配置
"""
DEBUG = True
class TestingConfig(BaseConfig):
"""
测试环境配置类，用于存放开发环境的配置
"""
DEBUG = True
TESTING = True
registered_app = [
'app'
]
config_map = {
'development': DevelopmentConfig,
'production': ProductionConfig,
'testing': TestingConfig
}
Что касается последнегоregistered_appиconfig_mapКакая польза? Вы можете сделать автоматический впрыск, о котором я расскажу позже.
Затем добавляю конфигурацию лога.Конфигурация лога очень важна.Разные команды разработчиков часто имеют набор стандартизированных шаблонов конфигурации лога, которые как правило не меняются, поэтому их можно определить прямо в коде или в виде конфигурационных файлов.
config/logger.py
from logging.config import dictConfig
def config_logger(enable_console_handler=True, enable_file_handler=True, log_file='app.log', log_level='ERROR',
log_file_max_bytes=5000000, log_file_max_count=5):
# 定义输出到控制台的日志处理器
console_handler = {
'class': 'logging.StreamHandler',
'formatter': 'default',
'level': log_level,
'stream': 'ext://flask.logging.wsgi_errors_stream'
}
# 定义输出到文件的日志处理器
file_handler = {
'class': 'logging.handlers.RotatingFileHandler',
'formatter': 'detail',
'filename': log_file,
'level': log_level,
'maxBytes': log_file_max_bytes,
'backupCount': log_file_max_count
}
# 定义日志输出格式
default_formatter = {
'format': '[%(asctime)s] %(levelname)s in %(module)s: %(message)s'
}
detail_formatter = {
'format': '[%(asctime)s] %(levelname)s in %(module)s: %(message)s'
}
handlers = []
if enable_console_handler:
handlers.append('console')
if enable_file_handler:
handlers.append('file')
d = {
'version': 1,
'formatters': {
'default': default_formatter,
'detail': detail_formatter
},
'handlers': {
'console': console_handler,
'file': file_handler
},
'root': {
'level': log_level,
'handlers': handlers
}
}
dictConfig(d)
Выше приведен типичный метод настройки журнала Python, который определяет переменные части как параметры (файл журнала, уровень и т. д.), определяет два обработчика журнала (файл и консоль) и требует только вызова этого метода при его использовании.
Класс приложения
После определения конфигурации мы можем приступить к созданию нашего приложения Flask. Студенты, которые использовали Flask, знают, что для создания приложения Flask требуется всего одна строка кода.
app = Flask(__name__)
Но это не готовый способ для производства.Для удобства производства и тестирования нам нужен метод для получения этого объекта приложения.
def create_app(conf=None):
# initialize logger
register_logger()
# check instance path
instance_path = os.environ.get('INSTANCE_PATH') or None
# create and configure the app
app = Flask(__name__, instance_path=instance_path)
if not conf:
conf = get_config_object()
app.config.from_object(conf)
# ensure the instance folder exists
if app.instance_path:
try:
os.makedirs(app.instance_path)
except OSError:
pass
# register app
register_app(app)
return app
Здесь было сделано несколько вещей, одна — зарегистрировать класс журнала, другая — загрузить объект конфигурации, а третья — создатьinstanceСправочник, четвертый, чтобы зарегистрировать бизнес-приложение.
Почему журнал регистрации находится на первой строке?
Многие разработчики выносят конфигурацию лога в класс конфигурации, это не большая проблема, но чем раньше вы зарегистрируете лог, тем быстрее начнет собираться ваш лог. Если журнал настроен после загрузки класса конфигурации, если при создании приложения сообщается об ошибке, он не может быть собран определенным нами сборщиком журналов.
Способ регистрации лога можно записать так
def register_logger():
log_level = os.environ.get('LOG_LEVEL') or 'INFO'
log_file = os.environ.get('LOG_FILE') or 'app.log'
config_logger(
enable_console_handler=True,
enable_file_handler=True,
log_level=log_level,
log_file=log_file
)
Я по-прежнему получаю конфигурацию из переменной среды и вызываю предыдущую функцию конфигурации для настройки журнала.
Метод для загрузки объекта конфигурации.
def get_config_object(env=None):
if not env:
env = os.environ.get('FLASK_ENV')
else:
os.environ['FLASK_ENV'] = env
if env in config.config_map:
return config.config_map[env]
else:
# set default env if not set
env = 'production'
return config.config_map[env]
отFLASK_ENVЭта переменная среды получает рабочую среду, а затем в соответствии с предыдущим классом конфигурацииconfig_mapПолучите соответствующий класс конфигурации и реализуйте загрузку класса конфигурации.
Последний шаг — зарегистрировать наш бизнес-код.
def register_app(app):
for a in config.registered_app:
module = importlib.import_module(a)
if hasattr(module, 'register'):
getattr(module, 'register')(app)
Это используется в классе конфигурацииregistered_appСписок, который определяет загружаемые модули. Для микросервисов обычно используется только один модуль.
мне все еще нужноapp/__init__.pyЕсть файлregisterметод, который выполняет определенные операции регистрации, такие как регистрация схемы Flask.
def register(app):
api_bp = Blueprint('api', __name__, url_prefix='/api')
app.register_blueprint(api_bp)
Зачем делать метод регистрации?
Поскольку у каждого бизнес-модуля есть своя маршрутизация, ORM или план и т. д., это собственный бизнес-код, который должен быть отделен от скелета. Использование конкретного метода в качестве спецификации удобно для расширения пользовательского кода, а другой метод прост для понимания командой и не требует гибкой настройки.Соглашение здесь больше, чем конфигурация. Конечно, у вас может быть другой набор собственных реализаций.
Я организовал приведенный выше код какapplication.pyмодуль
import os
import importlib
from flask import Flask
from config.logger import config_logger
from config import config
def register_logger():
log_level = os.environ.get('LOG_LEVEL') or 'INFO'
log_file = os.environ.get('LOG_FILE') or 'app.log'
config_logger(
enable_console_handler=True,
enable_file_handler=True,
log_level=log_level,
log_file=log_file
)
def register_app(app):
for a in config.registered_app:
module = importlib.import_module(a)
if hasattr(module, 'register'):
getattr(module, 'register')(app)
def get_config_object(env=None):
if not env:
env = os.environ.get('FLASK_ENV')
else:
os.environ['FLASK_ENV'] = env
if env in config.config_map:
return config.config_map[env]
else:
# set default env if not set
env = 'production'
return config.config_map[env]
def create_app_by_config(conf=None):
# initialize logger
register_logger()
# check instance path
instance_path = os.environ.get('INSTANCE_PATH') or None
# create and configure the app
app = Flask(__name__, instance_path=instance_path)
if not conf:
conf = get_config_object()
app.config.from_object(conf)
# ensure the instance folder exists
if app.instance_path:
try:
os.makedirs(app.instance_path)
except OSError:
pass
# register app
register_app(app)
return app
def create_app(env=None):
conf = get_config_object(env)
return create_app_by_config(conf)
предоставлено здесьcreate_app_by_configспособ создания непосредственно из класса конфигурацииappObject, в основном для облегчения прямого внедрения определенных классов конфигурации во время модульного тестирования.
Наш скелет в основном сформирован, включая самые основные классы конфигурации, настройку журнала и механизм регистрации приложений. Затем мы можем запустить наше приложение Flask.
тест на разработку
Колба обеспечиваетflask runкоманду для запуска тестового приложения, но также необходимо предоставитьFLASK_APPиFLASK_ENVДве переменные среды для начала, мы также можем упростить этот шаг.
написатьrun.py
import click
from envparse import env
from application import create_app
@click.command()
@click.option('-h', '--host', help='Bind host', default='localhost', show_default=True)
@click.option('-p', '--port', help='Bind port', default=8000, type=int, show_default=True)
@click.option('-e', '--env', help='Running env, override environment FLASK_ENV.', default='development', show_default=True)
@click.option('-f', '--env-file', help='Environment from file', type=click.Path(exists=True))
def main(**kwargs):
if kwargs['env_file']:
env.read_envfile(kwargs['env_file'])
app = create_app(kwargs['env'])
app.run(host=kwargs['host'], port=kwargs['port'])
if __name__ == '__main__':
main()
использовать здесьclickСоздан простой сценарий командной строки, который напрямую запускает тестовую службу с аргументами командной строки. Конечно, параметры по умолчанию доступны напрямую, используйтеpython run.pyИли щелкните правой кнопкой мыши в среде IDE для запуска. В то же время он также обеспечиваетenv-fileопция, пользователь может предоставить файл переменных среды.
Зачем использовать файл переменной среды?
Поскольку многие конфигурации производственной среды и среды разработки различаются, например общедоступные облачные ключи, подключения к базе данных и т. д., эту информацию нельзя отправлять в программное обеспечение для контроля версий, такое как git, поэтому мы можем создать файл .env следующим образом.
ACCESS_KEY=xxx
ACCESS_SECRET=xxx
Добавьте этот файл в gitignore, затем используйте--env-fileЗагрузка этого файла может быть использована непосредственно в среде разработки без необходимости каждый раз вводить его вручную.
развертывать
Мы точно не будем использовать тестовый метод для запуска рабочей среды, нам нужно что-то вродеgunicornПосле того, как инструмент запустит официальную службу, мы также можем использовать контейнерные технологии, такие как Docker, для автоматизации процесса производственного развертывания.
написатьserver.py
from application import create_app
app = create_app()
Здесь все очень просто, просто создайте объект приложения Flask, а затем вы можете передатьgunicorn server:appзапускать.
написатьrequirements.txtфайл для автоматической установки зависимостей. Более поздние зависимости могут быть записаны.
flask
flask-restful
click
envparse
gunicorn
написатьDockerfileдокумент
FROM python:3.8
COPY . /opt
WORKDIR /opt
RUN pip install --no-cache-dir -r requirements.txt
CMD ["gunicorn", "-b", "0.0.0.0:80", "server:app"]
Затем вы можете использовать следующую команду, чтобы запустить контейнер службы с помощью Docker.
docker build -t myapp:0.1 .
docker run -d --name myapp -p 80:80 myapp:0.1
На данный момент простой скелет Flask завершен, и вы можете увидеть полный проект ниже.
написать бизнес
Создание скелета Flask заняло минут 20. Для команды разработчиков скелет нужно разработать только один раз, а последующие проекты можно клонировать напрямую. Теперь давайте напишем конкретный сервис отправки СМС.
Какое общедоступное облако использовать?
В реальном бизнесе мы можем использовать одно облако или сочетание нескольких облаков. В нашем реальном бизнесе то, какой общедоступный облачный сервис использовать, зависит не от нас, а от того, у кого самая низкая цена, у кого больше скидок и у кого сильные функции. 😄
Таким образом, мы можем выделить общность SMS-бизнеса и написать абстрактный класс. Общие черты служб SMS в основном включают шаблоны SMS, подписи, получателей и параметры шаблона.
простой абстрактный класс
class SmsProvider:
def __init__(self, **kwargs):
self.conf = kwargs
def send(self, template, receivers, **kwargs):
pass
Затем идет реализация на базе Alibaba Cloud, следующий код модифицирован по официальному примеру
class AliyunSmsProvider(SmsProvider):
def send(self, template, receivers, **kwargs):
from aliyunsdkcore.request import CommonRequest
client = self.get_client(self.conf['app_key'], self.conf['app_secret'], self.conf['region_id'])
request = CommonRequest()
request.set_accept_format('json')
request.set_domain(self.conf['domain'])
request.set_method('POST')
request.set_protocol_type('https')
request.set_version(self.conf['version'])
request.set_action_name('SendSms')
request.add_query_param('RegionId', self.conf['region_id'])
request.add_query_param('PhoneNumbers', receivers)
request.add_query_param('SignName', self.conf['sign_name'])
request.add_query_param('TemplateCode', self.get_template_id(template))
request.add_query_param('TemplateParam', self.build_template_params(**kwargs))
return client.do_action_with_exception(request)
def get_template_id(self, name):
if name in self.conf['template_id_map']:
return self.conf['template_id_map'][name]
else:
raise ValueError('no template {} found!'.format(name))
@staticmethod
def get_client(app_key, app_secret, region_id):
from aliyunsdkcore.client import AcsClient
return AcsClient(app_key, app_secret, region_id)
@staticmethod
def build_template_params(**kwargs):
if 'params' in kwargs and kwargs['params']:
return json.dumps(kwargs['params'])
else:
return ''
затем вBaseConfigДобавьте следующую конфигурацию, которая является базовой конфигурацией некоторых API-интерфейсов общедоступного облака и должна загружаться через переменные среды во время работы, среди которыхtemplate_id_mapСодержимым в нем является название шаблона и соответствующий идентификатор, который используется для различения разных шаблонов SMS, таких как код подтверждения, акция и т. д., а имя используется в качестве параметра для вызывающего абонента, чтобы избежать прямой передачи Я БЫ.
# SMS config
SMS_CONF = {
'aliyun': {
'provider_cls': 'app.sms.AliyunSmsProvider',
'config': {
'domain': 'dysmsapi.aliyuncs.com',
'version': os.environ.get('ALIYUN_SMS_VERSION') or '2017-05-25',
'app_key': os.environ.get('ALIYUN_SMS_APP_KEY'),
'app_secret': os.environ.get('ALIYUN_SMS_APP_SECRET'),
'region_id': os.environ.get('ALIYUN_SMS_REGION_ID'),
'sign_name': os.environ.get('ALIYUN_SMS_SIGN_NAME'),
'template_id_map': {
'captcha': 'xxx'
}
}
}
}
Идентификатор шаблона, подпись, ключ приложения и секрет приложения необходимо получить из консоли Alibaba Cloud. Шаблон и подпись необходимо просмотреть, прежде чем их можно будет получить.
Таким же образом можно добавить API HUAWEI CLOUD, либо модифицировать их прямо из примера, но SDK у HUAWEI CLOUD пока нет, и его нужно вызывать через API, что аналогично.
class HuaweiSmsProvider(SmsProvider):
def send(self, template, receivers, **kwargs):
header = {'Authorization': 'WSSE realm="SDP",profile="UsernameToken",type="Appkey"',
'X-WSSE': self.build_wsse_header(self.conf['app_key'], self.conf['app_secret'])}
form_data = {
'from': self.conf['sender'],
'to': receivers,
'templateId': self.get_template_id(template),
'templateParas': self.build_template_params(**kwargs),
}
r = requests.post(self.conf['url'], data=form_data, headers=header, verify=False)
return r
def get_template_id(self, name):
if name in self.conf['template_id_map']:
return self.conf['template_id_map'][name]
else:
raise ValueError('no template {} found!'.format(name))
@staticmethod
def build_wsse_header(app_key, app_secret):
now = time.strftime('%Y-%m-%dT%H:%M:%SZ')
nonce = str(uuid.uuid4()).replace('-', '')
digest = hashlib.sha256((nonce + now + app_secret).encode()).hexdigest()
digest_base64 = base64.b64encode(digest.encode()).decode()
return 'UsernameToken Username="{}",PasswordDigest="{}",Nonce="{}",Created="{}"'.format(app_key, digest_base64,
nonce, now)
@staticmethod
def build_template_params(**kwargs):
if 'params' in kwargs and kwargs['params']:
return json.dumps(list(kwargs['params'].values()))
else:
return ''
Также добавьте конфигурацию, последнююBaseConfigкак показано ниже, гдеSMS_PROVIDERКонфигурация указанаSMS_CONF, указав, какой публичный облачный сервис мы сейчас используем:
class BaseConfig:
SECRET_KEY = os.environ.get('SECRET_KEY') or os.urandom(16)
DEBUG = False
TESTING = False
# SMS config
SMS_PROVIDER = os.environ.get('SMS_PROVIDER')
SMS_CONF = {
'aliyun': {
'provider_cls': 'app.sms.AliyunSmsProvider',
'config': {
'domain': 'dysmsapi.aliyuncs.com',
'version': os.environ.get('ALIYUN_SMS_VERSION') or '2017-05-25',
'app_key': os.environ.get('ALIYUN_SMS_APP_KEY'),
'app_secret': os.environ.get('ALIYUN_SMS_APP_SECRET'),
'region_id': os.environ.get('ALIYUN_SMS_REGION_ID'),
'sign_name': os.environ.get('ALIYUN_SMS_SIGN_NAME'),
'template_id_map': {
'captcha': 'xxx'
}
}
},
'huawei': {
'provider_cls': 'app.sms.HuaweiSmsProvider',
'config': {
'url': os.environ.get('HUAWEI_URL'),
'app_key': os.environ.get('HUAWEI_SMS_APP_KEY'),
'app_secret': os.environ.get('HUAWEI_SMS_APP_SECRET'),
'sender': os.environ.get('HUAWEI_SMS_SENDER_ID'),
'template_id_map': {
'captcha': 'xxx'
}
}
}
}
Аналогичным образом можно добавить и другие публичные облака.
Затем мы добавляем метод для получения одноэлементного объекта провайдера. Здесь мы используем объект g Flask, чтобы зарегистрировать наш объект Provider как глобальный одноэлементный объект.
from flask import g, current_app
from werkzeug.utils import import_string
def create_sms():
provider = current_app.config['SMS_PROVIDER']
sms_config = current_app.config['SMS_CONF']
if provider in sms_config:
cls = sms_config[provider]['provider_cls']
conf = sms_config[provider]['config']
sms = import_string(cls)(**conf)
return sms
return None
def get_sms():
if 'sms' not in g:
g.sms = create_sms()
return g.sms
После того, как все это будет сделано, вы можете добавить класс представления, который использует библиотеку Flask-Restful для создания представления API.
app/api/sms.py
import logging
from flask_restful import Resource, reqparse
from app.sms import get_sms
# 定义参数，参考 https://flask-restful.readthedocs.io/en/latest/reqparse.html
parser = reqparse.RequestParser(bundle_errors=True)
parser.add_argument('receivers', help='Comma separated receivers.', required=True)
parser.add_argument('template', help='Notification template name.', required=True)
parser.add_argument('params', help='Notification template params.', type=dict)
class Sms(Resource):
def post(self):
args = parser.parse_args()
sms = get_sms()
try:
res = sms.send(**args)
except Exception as e:
logging.error(e)
return {'message': 'failed'}, 500
if res.status_code < 300:
return {'message': 'send'}, 200
else:
logging.error('Send sms failed with {}'.format(res.text))
return {'message': 'failed'}, 500
Затем определяем маршрут.
app/api/__init__.py
from flask import Blueprint
from flask_restful import Api
from app.api.health import Health
from app.api.sms import Sms
api_bp = Blueprint('api', __name__, url_prefix='/api')
api = Api(api_bp)
api.add_resource(Sms, '/sms')
Наконец, не забудьте зарегистрировать план в нашем модуле приложения.
app/__init__.py
from app.api import api_bp
# register blueprint
def register(app):
app.register_blueprint(api_bp)
На этом наш микросервис SMS готов. Его можно протестировать и развернуть с помощью описанного выше метода.
Среди них мы определили некоторые переменные среды, которые можно загрузить через файл переменных среды во время тестирования и через переменные среды контейнера во время выполнения. Он помещается в каталог экземпляра, потому что экземпляр является нашим каталогом экземпляра Flask по умолчанию, и этот каталог не будет отправлен в git.
instance/env
SMS_PROVIDER=huawei
HUAWEI_URL=https://rtcsms.cn-north-1.myhuaweicloud.com:10743/sms/batchSendSms/v1
HUAWEI_SMS_APP_KEY=aaa
HUAWEI_SMS_APP_SECRET=bbb
HUAWEI_SMS_SENDER_ID=ccc
Загружается через переменные среды во время выполнения
docker run -d --name sms -p 80:80 \
-e "SMS_PROVIDER=aliyun" \
-e "ALIYUN_SMS_APP_KEY=aaa" \
-e "ALIYUN_SMS_APP_SECRET=bbb" \
-e "ALIYUN_SMS_REGION_ID=cn-hangzhou" \
-e "ALIYUN_SMS_SIGN_NAME=ccc" \
myapp:0.1
Полностью проект можно посмотреть здесь.
Затем мы можем выполнить следующий тест, обратите внимание на изменение идентификатора шаблона и переменных среды в конфигурации и изменение параметров в соответствии с нашими собственными параметрами шаблона.
Эпилог
Для старой птички может вообще не уйти 1 час на разработку этого проекта. Для канонических онлайн-проектов по-прежнему не хватает некоторых вещей, таких как модульные тесты. На что похож ваш производственный API-сервис? Обсуждения приветствуются!
Микросервис СМС здесь является лишь ориентиром, на самом деле все сервисы публичных облачных API можно применять одинаково. 1 час на запуск микросервиса, а остальные 7 часов на чистку наггетсов😄.
Я Хуояньцзюнь, пусть мое письмо рассеет одиночество души.