1 小時上線之用 Flask 開發一個短信微服務

時間 2020-02-08

原文原文鏈接

前言

Flask 是 Python 開發的輕量 Web 框架，有多輕量呢？10 行之內就能夠開發一個 Web 服務，不過這隻能用來作演示，今天我就用 1 個小時來開發一個用於生產環境的短信微服務。如下是咱們生產環境脫敏後直接可用的服務代碼，絕非示例教程。html

爲何要開發短信微服務？python

短信服務咱們都是依賴公有云的實現，經過公有云的 API 直接調用，那爲何還要本身封裝呢？git

由於微服務環境下咱們要減小代碼的重複量，若是有多個微服務須要使用短信服務，那就要複製多遍代碼，把公有云的 API 包裝成咱們本身的微服務 API 能夠將代碼的複製減小爲一行 Http 請求。
調用 API 的 accesskey 和 secret 不須要複製給多個服務，減小安全風險。
能夠根據咱們的業務需求加入共用的業務邏輯。

多了一層調用有沒有性能影響？github

多了一層調用是多了一個網絡請求，可是影響微乎其微。咱們不可能由於面向對象的方式太多調用就寫逐行執行的代碼吧。docker

公有云短信服務本就是異步調用，錯誤處理也是異步回調的方式。
微服務內部網絡的調用應該是很是快的，能夠同虛擬機部署或者同機房部署。

開始

首先咱們創建項目的骨架。數據庫

爲何要創建項目的骨架呢？json

由於 Flask 太過於輕量，因此例如配置、路由等規範須要由開發人員本身定義。通常成熟的開發團隊都有本身的一套開發骨架，要統一配置，統一開發規範，統一集成相關係統等。我這裏就分享一套適用於生產環境的很是簡單的開發骨架。flask

新建一個項目目錄，而後在裏面創建 app 和 config 兩個 Python 目錄。app 用於存放業務相關代碼，config 用於存放配置相關代碼。api

配置類

在 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__)
複製代碼

但這不是生產可用的方式，爲了生產和測試方便，咱們須要用一個方法獲取這個 app 對象。

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 目錄，四是註冊應用業務。

爲何註冊日誌要放在第一行？

很多開發人員會把日誌的配置放在配置類裏，這個沒太大問題，只是越早註冊日誌，你的日誌就會越早開始收集。若是載入配置類後才配置日誌，那若是建立 app 時報錯就沒法被咱們定義的日誌收集器收集到了。

註冊日誌的方法能夠這樣寫

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)
複製代碼

爲何要搞個 register 方法？

由於每一個業務模塊有本身的路由、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 方法用於從配置類直接建立 app 對象，主要是便於單元測試時直接注入特定的配置類。

咱們的骨架基本上就成型了，包括了最基礎的配置類、日誌配置和應用註冊機制。而後就能夠運行咱們的 Flask 應用了。

開發測試

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 app 對象便可，而後能夠經過 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 骨架就完成了，你們能夠在下面看到完整的項目。

Github Flask 骨架示例

編寫業務

上面大概用了 20 分鐘搞了個 Flask 的骨架，對於開發團隊來講，骨架只要開發一次，後續的項目直接克隆就好了。下面咱們就來編寫具體的發送短信業務。

使用哪一個公有云？

實際業務中咱們可能使用單一一個雲，也可能混合使用多個雲。在咱們的實際業務中，具體用哪一個公有云的服務，不是取決於咱們，而是取決於誰的價格低，誰的優惠多，誰的功能強。😄

因此咱們能夠提取短信業務的共性寫一個抽象類。短信服務的共同點主要有短信模版，簽名，接收人，模版參數等。

一個簡單的抽象類

class SmsProvider:

    def __init__(self, **kwargs):
        self.conf = kwargs

    def send(self, template, receivers, **kwargs):
        pass
複製代碼

而後有基於阿里雲的實現，如下代碼根據官方示例修改

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 裏的內容是模版的名稱和對應的 ID，用於區分不一樣的短信模版，如驗證碼，推廣等，名稱做爲參數供調用方使用，避免了直接傳遞 ID。

# 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'
                }
            }
        }
    }
複製代碼

其中模版 ID，簽名，App Key，App Secret 須要在阿里雲控制檯獲取，模版和簽名須要審覈後才能得到。

一樣的方法能夠添加華爲雲的 API，也可直接從示例修改，只是華爲雲暫時沒有 SDK，須要經過 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'
                }
            }
        }
    }
複製代碼

其餘的公有云也能夠經過相似的方式添加。

而後咱們添加一個方法，獲取 Provider 的單例對象。這裏使用 Flask 的 g 對象，把咱們的 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 模塊裏註冊藍圖。

app/__init__.py

from app.api import api_bp


# register blueprint
def register(app):
    app.register_blueprint(api_bp)
複製代碼

至此，咱們的短信微服務就完成了。能夠經過咱們上面的方法進行測試和部署。

其中咱們定義了一些環境變量，在測試時可經過環境變量文件載入，運行時可經過容器的環境變量載入。放在 instance 目錄下是由於 instance 是咱們默認的 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
複製代碼