python 工具函数代码
Posted 胖虎是只mao
tags:
篇首语：本文由小常识网(cha138.com)小编为大家整理，主要介绍了python 工具函数代码相关的知识，希望对你有一定的参考价值。
1. utc 标准时间转换成本地时间
def utc2local(utc_st):
'''将utc 标准时间转换成本地时间'''
now_stamp = datetime.now().timestamp()
local_time = datetime.fromtimestamp(now_stamp)
utc_time = datetime.utcfromtimestamp(now_stamp)
offset = local_time - utc_time
print(offset,'时区差') # 时区差
local_st = utc_st + offset
return local_st
# utc时间是在本初子午线
def date_str(date):
'''将 datetime 数据类型转换为本地时间后再转换成字符串返回'''
return utc2local(date).strftime('%Y-%m-%d %H:%M') if date else ''
time_stamp = 1525848792
utc_time = datetime.utcfromtimestamp(time_stamp)
temp = date_str(utc_time)
print(temp)
打印：
8:00:00 时区差
2018-05-09 14:53
2. 登陆、注册时的验证码生成和校验
import random
from flask import session
'''
有了 session 后，我们不仅可以用它来保持登录状态，还可以在用户登录前实现验证码功能。
思路很简单，只用两步：
第一步，用户进入登录页面时，我们生成一个验证码的问题和答案。然后，把答案保存到 session 中，把验证码问题传到前端页面显示。
第二步，当用户看到验证码问题，填完表单、输入验证码答案、点击提交表单按钮后，我们再在视图函数里面针对用户的这个 post 请求去对比 session 里的答案。
'''
# 生成验证码
def gen_verify_num():
a = random.randint(-20, 20)
b = random.randint(0, 50)
data =
'question': str(a) + ' + ' + str(b) + '= ?',
'answer' : str(a + b)
# 将答案保存到session中
session['ver_code'] = data['answer']
return data
# 校验验证码
def verify_num(code):
if code != session['ver_code']:
raise Exception('验证码输入错误！！！')
3. python对于mongo db数据库的读取和处理
mongo存储的数据在没有特别指定_id数据类型时，默认类型为ObjectID
‘_id’: ObjectId(‘55717c9eb2c983c127000000’)
python处理方式
基本思路就是转换成python对象 ,然后处理.
'''
这里新增了 _process_filter 函数和 get_list 函数，其中 _process_filter 函数用来转化 filter1 中 '_id' 对应的值，以便于数据库查询。
get_list 函数帮我们根据传入参数对数据库集合进行排序和数量限制，最终转化为列表返回。
'''
def _process_filter(filter1):
if filter1 is None:
return
_id = filter1.get('_id')
# 将传入参数 filter1 的 '_id' 对应的值转化为 ObjectId 类型
if _id and not isinstance(_id, ObjectId):
filter1['_id'] = ObjectId(_id)
def get_list(collection_name, sort_by=None, filter1=None, size=None):
_process_filter(filter1)
result = mongo.db[collection_name].find(filter1)
if sort_by:
result = result.sort(sort_by[0], sort_by[1])
if size:
result = result.limit(size)
result = list(result)
return result
python对象转换成mongo数据库中的ObjectID类型
from bson.objectid import ObjectId,InvalidId
from werkzeug.routing import BaseConverter, ValidationError
from itsdangerous import base64_encode, base64_decode
class ObjectIDConverter(BaseConverter):
def to_python(self, value):
try:
return ObjectId(value)
except (InvalidId, ValueError, TypeError):
raise ValidationError()
def to_url(self, value):
return str(value)
class Base64ObjectIDConverter(BaseConverter):
def to_python(self, value):
try:
return ObjectId(base64_decode(value))
except (InvalidId, ValueError, TypeError):
raise ValidationError()
def to_url(self, value):
return base64_encode(value.binary).decode('utf-8')
4. python 对于密码的处理，进行hash加密和密码对比
from werkzeug.security import generate_password_hash,check_password_hash
pa = generate_password_hash('abc123')
pa
'pbkdf2:sha256:50000$eENkoanm$da00f345a5d76fd19fa7e1e644cde62392d2290b86907813a9ed9deafabcc5cf'
check_password_hash(pa,'abc123')
True
check_password_hash(pa,'abc123421')
False
5. flask 异步发送邮件
import random
from flask import session, current_app
from flask_mail import Message
from threading import Thread
from . import extensions
‘’‘
send_mail_async 函数中，我们在应用上下文中调用 extensions.py 文件中的 mail 对象的 send 方法实现发送邮件功能
。send_email 中，我们使用了 current_app._get_current_object() 而不是 current_app 来获取当前应用实例，
这是因为这里的 app 对象需要传入其他线程中。
有了 app 后，我们就可以得到 app.config 里配置的信息。
然后，再把传入参数中的接收人和发送内容整合为 msg。接着开启多线程就行了
‘’’
# 设置接收者、发送内容等
def send_email(to, subject, body, is_txt=True):
app = current_app._get_current_object()
msg = Message(subject=app.config.get('MAIL_SUBJECT_PREFIX') + subject,
sender=app.config.get('MAIL_USERNAME'), recipients=[to])
if is_txt:
msg.body = body
else:
msg.html = body
# 多线程支持
thr = Thread(target=send_mail_async, args=[app, msg])
thr.start()
return thr
# 发送邮件
def send_mail_async(app, msg):
with app.app_context():
extensions.mail.send(msg)
6. 创建并初始化 Flask_app
def create_app():
""" 创建并初始化 Flask app
"""
app = Flask('rmon')
# 根据环境变量加载开发环境或生产环境配置
env = os.environ.get('RMON_ENV')
if env in ('pro', 'prod', 'product'):
app.config.from_object(ProductConfig)
else:
app.config.from_object(DevConfig)
# 从环境变量 RMON_SETTINGS 指定的文件中加载配置
app.config.from_envvar('RMON_SETTINGS', silent=True)
app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = False
# 注册 Blueprint
app.register_blueprint(api)
# 初始化数据库
db.init_app(app)
# 如果是开发环境则创建所有数据库表
if app.debug:
with app.app_context():
db.create_all()
return app
上面的代码中，分别导入了在前文中定义的蓝图 api 和 SQLAlchemy 对象 db 以及开发环境和生产环境配置类。在 create_app 函数中，首先创建了 Flask 应用，接着根据环境变量 RMON_ENV 的值来加载不同的配置文件。当 RMON_ENV 环境变量的值是 product 时则加载 ProductConfig 类对应的配置项，否则加载 DevConfig 类对应的配置项。最后注册了 api 蓝图，并对 db 对象进行初始化配置。由于我们使用了 flask_sqlalchemy 扩展，所以在执行 db.init_app(app) 代码时，会自动从配置文件中查找 SQLALCHEMY_DATABASE_URI 配置项进行配置数据库地址。
在 created_app 函数的最后，当发现打开了 debug 模式（app.debug == True）时，会自动通过 db.create_all() 方法创建所有的数据库表。这是因为当应用加载的是 DevConfig 配置文件时，应用将处于开发模式且 app.debug 属性为 True，这个时候配置的数据库地址是内存数据库，可以调用方法自动创数据库。如果是在生产环境中，不是配置的内存数据库，则这段代码绘不会执行也不应该执行。
7. Flask 应用加载json配置文件
说明： 在 Flask 应用中常常需要加载各种配置信息，在 许多 项目中，我们通过 config.py 文件提供配置信息。但很多情况下，我们希望从其它格式的配置文件中加载配置，例如 JSON 文件。
尝试在 Flask 应用中加载 JSON 配置文件。
准备工作：
需要在 ~/Code/app.py 文件中编写一个创建 Flask 应用的函数 create_app 。创建 Flask 应用后从环境变量 RMON_CONFIG 指定的 JSON 文件中读取配置，并将配置信息存储在 Flask 应用的 config 属性中。
示例代码如下：
import os
from flask import Flask
def create_app():
""" 创建并初始化 Flask app
Returns:
app (object): Flask App 实例
"""
app = Flask('rmon')
# 获取 json 配置文件名称
file = os.environ.get('RMON_CONFIG')
# TODO 从 json_file 中读取配置项并将每一项配置写入 app.config 中
return app
提示：
JSON 配置文件不是标准的 JSON 文件，其中可能包含以 # 开头的，或者多个空格和 # 开头的注释行，注释在处理时应该被丢弃，不应该被当做配置项，示例配置文件如下：
# 这是注释行
# 这是注释行
"SQLALCHEMY_URI": "sqlite:///"
Python 中可以通过标准软件包 json 处理 JSON 文件
代码实现：
import os
import json
from flask import Flask
def create_app():
app = Flask('rmon')
file = os.environ.get('RMON_CONFIG')
try:
with open(file) as f:
data = ''
for line in f:
if not line.strip().startswith('#'):
data += line
config_dict = json.loads(data)
for key, value in config_dict.items():
app.config[key.upper()] = value
except json.decoder.JSONDecodeError:
print('配置文件读取数据有误')
except TypeError:
print('配置文件不存在')
return app
8. 捕捉异常，并触发自定义异常
class RestException(Exception):
"""异常基类
"""
def __init__(self, code, message):
"""初始化异常
Aargs:
code (int): http 状态码
message (str): 错误信息
"""
self.code = code
self.message = message
super(RestException, self).__init__()
代码非常简单，基于 Exception 实现了 RestException 异常基类。该异常有两个初始化参数，其中 code 代码发生异常时应该返回给客户端的 HTTP 状态码，message 为错误信息。有了 RestException 后，我们接着实现 ping 方法：
from rmon.common.rest import RestException
# 省略部分代码
class Server(db.Model):
# 省略部分代码
def ping(self):
"""检查 Redis 服务器是否可以访问
"""
try:
return self.redis.ping()
except RedisError:
raise RestException(400,
'redis server can not be connected'.format(self.host))
调用了 redis.StrictRedis 对象的 ping 方法用于检测 Redis 服务器是否可以连接，如果异常发生则捕获异常，并抛出自定义的 RestException 异常，其中指定了 HTTP 状态码 为 400。
9. redis 服务器模型的序列化和反序列化——使用Marshmallow 软件包
实现了 Server 代表的 Redis 服务器模型，主要代码如下：
from flask_sqlalchemy import SQLAlchemy
from datetime import datetime
db = SQLAlchemy()
class Server(db.Model):
"""Redis服务器模型
"""
__tablename__ = 'redis_server'
id = db.Column(db.Integer, primary_key=True)
# unique = True 设置不能有同名的服务器
name = db.Column(db.String(64), unique=True)
description = db.Column(db.String(512))
host = db.Column(db.String(15))
port = db.Column(db.Integer, default=6379)
password = db.Column(db.String())
updated_at = db.Column(db.DateTime, default=datetime.utcnow)
created_at = db.Column(db.DateTime, default=datetime.utcnow)
# 省略其它代码
# ...
比如在获取 Redis 服务器列表 API 中，会返回所有的 Redis 服务器对应的 Server 数据库模型对象的 json 对象数据。将 Server 实例转换为 json 对象的过程称为序列化。而在创建 Redis 服务器的 API 中，需要将客户端发送的 json 对象转换为 Server 实例，这个过程被称为反序列化。
对于 Server 实例转换为 json 对象的操作，比较容易想到的方式是通过 Python 标准软件包的 json.dump 方法进行，但这样会抛出 TypeError: Object of type ‘Server’ is not JSON serializable 异常。这是因为 Server 类型的对象默认不支持 json 序列化操作。并且在反序列化的过程中我们希望对数据进行验证，例如将 json 对象反序列化为 Server 实例时，希望 Server.host 对应的数据必须是有效的 IP 地址，否则反序列化失败。为了满足这些需求，我们使用 Marshmallow 软件包。 Marshmallow 的工作方式是定义一个对应于 Server 的序列化类 ServerSchema ，后面的所有序列化和反序列化工作都基于 ServerSchema 类进行。
代码如下：
from marshmallow import (Schema, fields, validate, post_load,
validates_schema, ValidationError)
# 省略了 Server 类的实现代码
# marshmallow 是用来实现复杂的 ORM 对象与 Python 原生数据类型相互转换的库
# 例如 Server 映射类实例与 JSON 对象相互转换
# 相互转换需要一个中间载体，这个中间载体就是 Schema 子类的实例
# 继承基类 Schema 创建子类，属性对应 Server 类的属性，每个字段都有一些约束
# Server 实例转换为字典或 JSON 字符串的过程叫做序列化
# JSON 字符串或字典转换为 Server 实例的过程叫做反序列化
# 序列化和反序列化需要对某些字段进行验证，使用 marshmallow 可以很好地实现需求
class ServerSchema(Schema):
"""Redis服务器记录序列化类
"""
# 序列化是将 Server 实例作为参数调用 dump 或 dumps 方法
# 返回一个字典对象或 JSON 字符串
# 反序列化是将字典对象或 JSON 字符串作为参数调用 load 或 loads 方法
# 返回一个 Server 实例
# dump_only 表示该字段只能序列化，load_only 表示该字段只能反序列化
id = fields.Integer(dump_only=True)
name = fields.String(required=True, validate=validate.Length(2, 64))
description = fields.String(validate=validate.Length(0, 512))
# host 必须是 IP v4 地址，通过正则验证
host = fields.String(required=True,
validate=validate.Regexp(r'^\\d1,3\\.\\d1,3\\.\\d1,3\\.\\d1,3$'))
port = fields.Integer(validate=validate.Range(1024, 65536))
password = fields.String()
updated_at = fields.DateTime(dump_only=True)
created_at = fields.DateTime(dump_only=True)
# 使用
装饰器创建验证器函数，函数名自定义
# 该方法在 self 调用 load 或 loads 方法时自动运行
# 也就是在添加和更新 Server 实例时会运行
@validates_schema
def validate_schema(self, data):
"""验证是否已经存在同名 Redis 服务器
"""
if 'port' not in data:
data['port'] = 6379
# ServerSchema 继承了 Schema 类，而 Schema 继承了 BaseSchema 类
# BaseSchema.__init__ 方法有默认参数 context ，它的默认值是 None
# 且 BaseSchema.__init__ 中定义了实例属性 context 的默认值为空字典
# 如果服务器已经存在，在相关请求出现时
# self.context 的 instance 字段值会被定义为 Server 实例
instance = self.context.get('instance', None)
server = Server.query.filter_by(name=data['name']).first()
# 创建服务器时，server 的值为 None，验证完毕
# 更新服务器时，如果更新服务器的 name ，server 的值也是 None ，验证完毕
if server is None:
return
# 创建服务器时，反序列化时不允许创建已经存在的服务器
if instance is None:
raise ValidationError('Redis server already exist', 'name')
# 更新服务器时，instance 应该为 Server 实例
if server != instance:
# 如果更新服务器的名字时用了另一个服务器的名字，会触发此异常
raise ValidationError('Redis server already exist', 'name')
# 在反序列化时，如果通过了验证器的验证，则自动运行此方法
@post_load
def create_or_update(self, data):
"""数据加载成功后自动创建 Server 对象
"""
instance = self.context.get('instance', None)
# 添加服务器时，instance 是 None
# 更新服务器时，instance 是 Server 实例
# 创建 Redis 服务器
if instance is None:
return Server(**data)
# 更新服务器
for key in data:
setattr(instance, key, data[key])
return instance
上面的代码中，省略了前面章节中已经实现的部分代码。通过继承来自于 Marshmallow 软件包的 Schema 基类实现了 ServerSchema 类，可以看到 ServerSchema 的字段对应于 Server 的属性，且每一个字段都有一些约束，比如 host 字段，设置了 required=True 则要求反序列化时 host 字段对应的数据必须存在，而且通过 validate 设置了格式必须符合 IP 地址正则表达式。在使用 ServerSchema 进行反序列化时，不允许创建已存在的 Redis 服务器，所以需要对 json 数据进行检查，如果发现已经有同名的 Redis 服务器则反序列化失败。上面的代码中， 通过 marshmallow.validates_schema 装饰器装饰的 validate_schema 方法完成了这样的功能需求。需要注意的是，实际上在创建 Redis 服务器和更新服务器时都需要验证是否有同名的服务器存在，为了区分是创建和更新操作，我们使用了 marshmallow.Schema 的 context 属性，如果当发现该属性中已存在 Server 对象时，则表明这是一个更新操作，否则是创建操作。当发现有同名 Server 对象存在时就抛出 ValidationError 异常。
我们还使用了 marshmallow.post_load 装饰器，这个装饰器的用途是设置 json 字符串被成功加载后执行的方法。在这里，我们设置了 json 字符串被成功加载后就创建或更新 Server 对象。这样加载 json 字符串后，就得到了新的 Server 对象。
10. flask 实现数据转换视图控制器基类，并捕获异常工具类
实验中定义的 API 接口必须返回 json 编码的字符串数据，那该怎么操作呢？我们可以在每一个 API 中都编写一遍序列化这些对象数据的代码，但是显然这样代码复用程度太低。第一节实验中的 IndexView 首页视图控制器是通过继承 flask.views.MethodView 实现的，所以不难想到能否实现一个类似于 flask.views.MethodView 的视图控制器基类，在基类中完成所有的数据转换操作。查看 MethodView 的源代码，发现其核心方法是 dispatch_request，内容如下：
class MethodView(with_metaclass(MethodViewType, View)):
'''注释
'''
def dispatch_request(self, *args, **kwargs):
meth = getattr(self, request.method.lower(), None)
# If the request method is HEAD and we don't have a handler for it
# retry with GET.
if meth is None and request.method == 'HEAD':
meth = getattr(self, 'get', None)
assert meth is not None, 'Unimplemented method %r' % request.method
return meth(*args, **kwargs)
看到 dispatch_request 的工作原理很简单，只需要调用和 HTTP 请求方法同名的视图控制器方法即可，获取方法通过 meth = getattr(self, request.method.lower(), None) 实现，最后直接返回视图控制器方法的执行结果。所以容易想到，可以通过继承 MethodView 实现一个新的视图控制器基类，并重新实现 dispatch_request 方法，然后获取视图控制器方法的执行结果也就是 meth(*args, **kwargs) 的执行结果，但并不直接返回其结果，而是对其进行序列化操作，并将序列化的结果转换成字符串数据生成 HTTP 响应对象后返回，这样就满足了 API 返回数据格式的需求。同时，在调用视图控制器方法时，可以捕获前文中定义的 RestExcepetion 异常，这样就能够保持控制器方法的简洁性。
经过上面的分析，可以实现 RestView 视图控制器基类
'''
该模块实现了 restful 的相关类型，主要包含 RestException 和 RestView 两个类型
RestException 是 restful 类型的异常基类
该类型的异常发生时，将被自动序列化为 JSON 格式响应
RestView 实现了 restful 视图基类
基于该基类实现视图控制器时，执行结果将被序列化为 JSON 格式响应
'''
from collections import Mapping
from flask import request, Response, make_response
from flask.json import dumps
from flask.views import MethodView
class RestException(Exception):
# 省略已完成的代码
# 继承 MethodView 类创建新的视图方法类
# 该类中定义的全部方法会被其子类继承并在子类中被调用
# 所有的 API 都将基于这个 RestView 类实现
# 当客户端发送请求到服务器，会调用此类中的方法
class RestView(MethodView):
"""自定义 View 类
JSON 序列化，异常处理，装饰器支持
"""
content_type = 'application/json; charset=utf-8'
method_decorators = []
# 如果遇到报错，例如 RestException ，就调用此方法
# 这是在 dispatch_request 方法中设置的
def handler_errorpython 工具函数代码