preface
Flask is a lightweight Web framework for Python development. How light is it? You can develop a Web service in less than 10 lines, but this is for demonstration purposes only, so today I spent an hour developing an SMS microservice for a production environment. The following is not a sample tutorial, but a service code that is directly available after desensitization in our production environment.
Why to develop SMS micro service?
SMS service we are dependent on the implementation of public cloud, through the API of the public cloud directly call, then why to encapsulate their own?
- Because in the microservice environment, we need to reduce the amount of code duplication. If multiple microservices need to use SMS service, we need to copy the code multiple times. Wrapping the API of public cloud into our own microservice API can reduce the code duplication to one line of Http request.
- Accesskey and Secret calls to the API do not need to be copied to multiple services, reducing security risks.
- Common business logic can be added based on our business needs.
Is there any performance impact of having an extra layer of calls?
An extra layer of calls is an extra network request, but the impact is minimal. We can’t write line-by-line code just because there are too many calls in the object-oriented way.
- The public cloud SMS service is called asynchronously, and error handling is also called asynchronously.
- The invocation of the microservice internal network should be very fast and can be deployed with a virtual machine or with a machine room.
start
First we build the skeleton of the project.
Why build the skeleton of a project?
Flask is so lightweight that specifications such as configuration, routing, and so on need to be defined by the developers themselves. Generally mature development teams have their own development framework, unified configuration, unified development specifications, unified integration of relevant systems, etc. I’m going to share a very simple development skeleton for a production environment.
Create a new project directory and create the app and config Python directories inside. App stores service codes, and config stores configuration codes.
The configuration class
Add the following to config/config.py, configuration design varies from person to person, Flask does nothing to limit it. My design here is to use BaseConfig as the base configuration class to store all the common configurations, while different environments use different subclasses of configuration that only need to change specific values for easy viewing.
If the configured value needs to be injected at run (such as a database connection, etc.), then environment variables can be used in the manner (SECRET_KEY below). I also use OR to provide a default value without environment variables.
import os
class BaseConfig:
"" Configuration base class for common configuration
SECRET_KEY = os.environ.get('SECRET_KEY') or os.urandom(16)
DEBUG = False
TESTING = False
class ProductionConfig(BaseConfig):
Production environment configuration class, used to store production environment configuration
pass
class DevelopmentConfig(BaseConfig):
Development environment configuration class, used to store development environment configuration.
DEBUG = True
class TestingConfig(BaseConfig):
Test environment configuration class for storing the configuration of the development environment
DEBUG = True
TESTING = True
registered_app = [
'app'
]
config_map = {
'development': DevelopmentConfig,
'production': ProductionConfig,
'testing': TestingConfig
}
Copy the codeWhat does registered_app and config_map do after that? You can do auto-injection, which I’ll talk about later.
Then I add a log configuration. Log configuration is very important. Different development teams often have a standard log configuration template that does not change, so it can be defined directly in the code, or in a configuration file.
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):
Define the log handler for output to the console
console_handler = {
'class': 'logging.StreamHandler'.'formatter': 'default'.'level': log_level,
'stream': 'ext://flask.logging.wsgi_errors_stream'
}
# define the log handler to output to the file
file_handler = {
'class': 'logging.handlers.RotatingFileHandler'.'formatter': 'detail'.'filename': log_file,
'level': log_level,
'maxBytes': log_file_max_bytes,
'backupCount': log_file_max_count
}
# define log output format
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)
Copy the codeThe above is a typical Python log configuration method that defines variable parts as parameters (log files, levels, and so on), defines two log handlers (file and console), and only calls this method when used.
Application class
With the configuration defined, we can start creating our Flask application. For those of you who have used Flask, you only need one line of code to create a Flask application.
app = Flask(__name__)
Copy the codeBut this is not the way to do it in production. For production and testing purposes, we need a way to get this app object.
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
Copy the codeHere we do several things, one is to register the logging class, two is to load the configuration object, three is to create the instance directory, four is to register the application service.
Why is the registration log in the first line?
Many developers will put logging configuration in the configuration class. This is not a problem, but the sooner you register logging, the sooner your logs will be collected. If logging is configured after the configuration class is loaded, it will not be collected by the logging collector we defined if an error occurs when creating the app.
The method for registering logs can be written like this
def register_logger(a):
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
)
Copy the codeI still get the configuration from the environment variable and call the previous configuration function to configure the log.
Method to load a configuration object.
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]
Copy the codeGet the running environment from the FLASK_ENV environment variable and then load the configuration class according to config_map in the previous configuration class.
The last step is to register our business code.
def register_app(app):
for a in config.registered_app:
module = importlib.import_module(a)
if hasattr(module, 'register'):
getattr(module, 'register')(app)
Copy the codeThis uses the registered_app list in the configuration class, which defines the module to load. For microservices, there is usually only one module.
I also need a register method in the app/__init__.py file to perform specific registration operations, such as registering the Flask blueprints.
def register(app):
api_bp = Blueprint('api', __name__, url_prefix='/api')
app.register_blueprint(api_bp)
Copy the codeWhy do we have a register method?
Because each business module has its own routing, ORM, or blueprint, etc., this is the business’s own code and must be decoupled from the skeleton. Using a specific method as a specification is (1) easy to customize the code extension, (2) easy to understand by the team, does not require flexible configuration, where convention is greater than configuration. Of course you can have another implementation of your own.
I’ve organized the above code into the application.py module
import os
import importlib
from flask import Flask
from config.logger import config_logger
from config import config
def register_logger(a):
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)
Copy the codeThe create_app_by_config method is provided to create the app object directly from the configuration class, mainly so that specific configuration classes can be injected directly during unit testing.
We basically have the skeleton in place, including the basic configuration classes, logging configuration, and application registration mechanism. Then we can run our Flask application.
Develop test
Flask provides the Flask run command to run the test application, but FLASK_APP and FLASK_ENV need to be provided to start the test application.
Write the 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()
Copy the codeHere you create a simple command-line script with Click to launch a test service directly with command-line arguments. Of course, the default parameters can be used directly, either by using Python run.py or by right-clicking in the IDE. The env-file option is also provided, allowing the user to provide a file of environment variables.
Why use environment variable files?
Because many configurations of production and development environments are different, such as public cloud keys, database connections, etc., this information should never be submitted to git or other version control software, so we can create a.env file as follows
ACCESS_KEY=xxx
ACCESS_SECRET=xxx
Copy the codeAdd the file to gitignore and load it with –env-file to be used in the development environment instead of typing it manually each time.
The deployment of
For production environment, we will definitely not start the production environment in the way of test. We need tools like Gunicorn to start a formal service. We can also use container technology like Docker to automate the production deployment process.
Write server. Py
from application import create_app
app = create_app()
Copy the codeThis is easy, just create a Flask app object, which can then be launched via gunicorn Server :app.
Write a requirements.txt file for automatically installing dependencies. You can write down the dependencies you use later.
flask
flask-restful
click
envparse
gunicorn
Copy the codeWrite a Dockerfile file
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"]
Copy the codeYou can then start the service container with Docker using the following command.
Docker build-t myapp: 0.1. docker run-d --name myapp -p 80:80 myapp:0.1
Copy the codeAt this point, a simple Flask skeleton is complete, and you can see the full project below.
Github Flask skeleton example
Writing business
Flask’s skeleton was built in about 20 minutes. For the development team, the skeleton only needs to be developed once and then cloned for subsequent projects. Let’s write a specific SMS service.
Which public cloud to use?
In real business we may use a single cloud or a mix of clouds. In our actual business, the specific use of public cloud services does not depend on us, but depends on whose price is low, who has more concessions, and who has strong functions. 😄
So we can extract the common message service to write an abstract class. The common features of SMS services include SMS template, signature, receiver, template parameters and so on.
A simple abstract class
class SmsProvider:
def __init__(self, **kwargs):
self.conf = kwargs
def send(self, template, receivers, **kwargs):
pass
Copy the codeThen there is the implementation based on Ali Cloud. The following code is modified according to the official example
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 ' '
Copy the codeThen add the following configuration to BaseConfig, which is the basic configuration of some public cloud APIS. It needs to be loaded by environment variables at runtime. Template_id_map contains the name and ID of the template, which is used to distinguish between different SMS templates, such as verification code, promotion, etc. The name is used as a parameter by the caller, avoiding passing the ID directly.
# 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'}}}}Copy the codeAmong them, template ID, signature, App Key and App Secret need to be obtained from Ali Cloud console, and template and signature can only be obtained after review.
In the same way, you can add the HUAWEI cloud API or modify it from the example. However, huawei cloud does not have the SDK, so you need to invoke it through the 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(The '-'.' ')
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 ' '
Copy the codeThe final BaseConfig is shown below, where the SMS_PROVIDER configuration specifies the SMS_CONF key and specifies which public cloud service we are using:
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'}}}}Copy the codeOther public clouds can be added in a similar manner.
We then add a method to get the Provider’s singleton object. Flask g objects are used here to register our Provider object as a global singleton.
from flask import g, current_app
from werkzeug.utils import import_string
def create_sms(a):
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(a):
if 'sms' not in g:
g.sms = create_sms()
return g.sms
Copy the codeOnce you’ve done that, you can add a view class that uses the Flask-restful library to generate API views.
app/api/sms.py
import logging
from flask_restful import Resource, reqparse
from app.sms import get_sms
# define parameters, refer to 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
Copy the codeThen we define the route.
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')
Copy the codeFinally, register the blueprints in our app module.
app/__init__.py
from app.api import api_bp
# register blueprint
def register(app):
app.register_blueprint(api_bp)
Copy the codeAt this point, our SMS micro service is completed. This can be tested and deployed in the same way we did above.
We defined some environment variables that can be loaded from the environment variables file at test time and from the container’s environment variables at run time. Instance is in git because instance is our default Flask instance directory, which is not committed to 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
Copy the codeThe runtime is loaded by environment variables
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.1Copy the codeThe full project can be viewed here.
Sample project code
We can then do the following tests, paying attention to changing the template ID and environment variables in the configuration, and modifying params based on our template parameters.
conclusion
For the old bird, it may not take an hour to develop this project. For standard online projects, there are still some things missing, such as unit tests. What are your production API services like? Welcome to the discussion!
The short message micro-service here is just a primer. In fact, all public cloud API services can be used in the same way. 1 hour online a micro service, the remaining 7 hours to brush gold 😄.
I am the fire eye gentleman, may my writing dispel the loneliness of the heart.
reference
- palletsprojects.com/p/flask/
- flask-restful.readthedocs.io/en/latest/
- Github.com/wwtg99/flas…
- Ali Cloud SMS API
- Huawei cloud SMS API
- Github.com/wwtg99/sms-…
- zhuanlan.zhihu.com/p/104380919