Flaskz介绍4--常用函数

为提高开发效率，Flaskz提供了一些常用的工具类函数，下面进行分类介绍。

系统路径&配置

• get_app_path(*path)
  • 功能：获取应用文件路径
  • 参数：
    • *path --需要连接的一个或多个路径列表(关键字参数)
  • 返回：应用文件路径
  • 示例：

                                   
                                        get_app_path('./_log', 'syslog.txt') # 获取_log目录下的syslog.txt文件
                                    
                                

• init_app_config(app)
  • 功能：初始化应用配置信息，初始化了以后，就可以通过get_app_config方法获取系统配置项，主要目的是方便使用
  • 参数：
    • app --可以是Flask应用/Config配置类/配置字典
  • 示例：

                                   
                                        init_app_config(app)    # Flask
                                        init_app_config(DevelopmentConfig)  # Class
                                        init_app_config({...})  # dict
                                    
                                

• get_app_config(key=None)
  • 功能：获取当前应用的指定配置信息
  • 参数：
    • key --配置属性名，如果为None，返回所有配置项dict
  • 返回：指定的应用配置/全部配置项
  • 示例：

                                   
                                        get_app_config('FLASKZ_LOGGER_FILENAME') # 获取为系统日志配置的文件名
                                    
                                

数据缓存

可以把数据在当前应用current_app和请求g应用上下文对象上进行缓存，从而提高数据的访问速度。

• set_app_cache(key, data, expire_minutes=0)
  • 功能：在当前应用current_app上缓存数据，一般用于缓存固定不会发生变化的数据
  • 参数：
    • key --缓存名称
    • data --缓存的值
    • expire_minutes --过期时间，单位是分钟，>0时起作用，过期以后通过get_app_cache获取到的数据为None
  • 示例：

                                   
                                        set_app_cache('sys_menus', menus, expire_minutes=60) # 缓存所有菜单信息，并在60分钟以后过期
                                    
                                

• get_app_cache(key)
  • 功能：获取当前应用上的缓存值
  • 参数：
    • key --要获取缓存值的key名称
  • 返回：当前应用上指定的缓存值，如果没有对应的键值或数据已经过期，则返回None
  • 示例：

                                   
                                        get_app_cache('sys_menus') # 获取缓存的菜单信息
                                    
                                

• clear_app_cache()
  • 功能：清空当前应用上所有的缓存值，一般用于数据发生变化以后重新加载缓存数据，比如在数据库中加了一个新的菜单
  • 示例：

                                   
                                        clear_app_cache()
                                    
                                

• set_g_cache(key, data)
  • 功能：在当前g应用上下文上设置缓存数据，整个请求处理过程g上的数据可以共享，每次请求都会重置g对象，一般存储数据库session之类的属性
  • 参数：
    • key --缓存名称
    • data --缓存的值
  • 示例：

                                   
                                        set_g_cache('db_session', DBSession()) # 将session缓存到g上，整个请求处理过程中只需要生成一次session即可
                                    
                                

• get_g_cache(key)
  • 功能：获取当前g应用上下文上的缓存值
  • 参数：
    • key --要获取缓存值的key名称
  • 返回：前g应用上下文上的缓存值，如果没有，则返回None
  • 示例：

                                   
                                        def get_db_session():
                                            """
                                            Get the db session from g.
                                            If not exist, create a session and return.
                                            :return:
                                            """
                                            session = get_g_cache('db_session') # 获取数据库session
                                            if session is None:
                                                session = DBSession()
                                                set_g_cache('db_session', session) # 如果没有则生成并缓存，供其他函数调用
    
                                            return session
                                    
                                

• remove_g_cache(key)
  • 功能：移除g应用上下文上设置的缓存数据，一般不需要调用，因为请求处理完成以后，flask会重置g变量
  • 参数：
    • key --要移除键值的名称

周期&定时执行

• set_timeout(interval, function, args=None, kwargs=None, daemon=True)
  • 功能：该定时器在定时器到期后执行一个函数或指定的一段代码
  • 参数：
    • interval --定时器在执行指定的函数或代码之前应该等待的时间，单位是毫秒
    • function --当定时器到期后，将要执行的函数
    • args --要执行的函数的位置参数
    • kwargs --要执行的函数的关键字参数
    • daemon --如果为True，计时器将是守护进程（当只剩下守护线程时，整个Python程序退出）
  • 示例：

                                   
                                       t = set_timeout(10, print, ('Hello, World!',))   # 10ms后调用print函数打印'Hello, World!'
                                       # t.cancel()    # 取消定时器
                                    
                                

• set_interval(interval, function, args=None, kwargs=None, immediately=False, daemon=True)
  • 功能：以指定的时间间隔（毫秒）周期性地执行一个函数。如果函数执行结果为False，则循环中断。
  • 参数：
    • interval --每次间隔的毫秒数
    • function --要重复调用的函数
    • args --要执行的函数的位置参数
    • kwargs --要执行的函数的关键字参数
    • immediately --如果为True，则会立即执行函数，如果为False，则会在interval间隔以后执行
    • daemon --如果为True，计时器将是守护进程（当只剩下守护线程时，整个Python程序退出）
  • 示例：

                                   
                                        t = set_interval(10, print, ('Hello, World!',)) # 每10ms调用一次print函数打印'Hello, World!'
                                        # t.cancel()    # 取消周期性运行
                                    
                                

  请注意，间隔是从函数执行完开始计时，而不是等间隔

  

工具类&对象操作

• Attribute

  Attribute是为了方便属性存储和访问提供的工具类，可以通过关键字参数任意存放和访问Attribute实例中的数据。

  和字典对象的区别就是，Attribute实例的属性值是通过"."进行访问的，而字典对象通过get方法进行访问。

                         
                              attr = Attribute(**{
                                      'start_time': start_time,
                                      'end_time': end_time,
                                      'duration': (end_time - start_time) * 1000,  # ms
                                      'statement': statement,
                                      'parameters': parameters,
                                      'context': context
                              })
                              print(attr.start_time)      # 通过.进行访问
                              print(attr.statement)
                              print(attr.any_key)         # 可以访问任意属性，而不会报错
                          
                      

• ins_to_dict(ins, option=None)
  • 功能：将数据模型类实例对象中的属性值转换为字典对象，一般用于数据交互和序列化传输
  • 参数：
    • ins --要进行转换的类实例对象，比如查询到的User数据模型类对象
    • option --转换选项，级联关系对象相关的转换选项需要以平铺的模式进行设置，有如下选项

      选项	类型	说明	实例
      cascade	int	需要转换的relationship关系层级数， 用于级联数据的转换，默认只转换当前实例对象中的值， 例如：User模型类中包含addresses关系列表，cascade=1，则会将关系中的Address对象也转换为字典，并附加到user字典对象上 请注意，如果对象之间有循环引用，则会在循环引用处中断转换	'cascade':1 # 只对第一层级联关系进行转换 'cascade':3 # 向下钻取，对第三层及以内的级联关系进行转换
      relationships	list	需要转换的relationship关系列表， 用于控制级联数据是否转换， 例如：User模型类中包含addresses和role两个关系，如果relationships=['role']，则只会将关系中的Role对象转换为字典，并附加到user字典对象上 相较于cascade，relationships可以更加精细的控制relationship关系是否转换输出 如果cascade>=1,则模型类所有的relationship关系都会被转换输出 如果cascade==0，relationships=['role']，则只有role对应的relationship被转换输出	'relationships':['role'] # 只对role级联关系进行转换
      include	list/function	指定需要转换的属性列表或过滤函数	'include':["id","name"] #只有id和name属性会输出到字典中 'include':lambda ins, key: ins.__class__.to_dict_field_filter(key) # 通过to_dict_field_filter函数进行过滤，只有过滤器返回为True的属性才会转换
      exclude	list/function	指定不需要转换的属性列表或过滤函数 请注意，如果同时指定了include和exclude，会先进行include检测，再进行exclude检测	'exclude':["password","email"], # password和email属性不会输出到字典中 'exclude':lambda ins, key: key in ["password","email"] # 通过函数进行过滤，如果过滤器返回为True的属性，则不会转换
      getattrs	function	指定如何获取实例对象所有的属性集合， 如果不指定，则会从对象的__dict__中获取， ModelMixin扩展类，默认从get_to_dict_attrs方法中返回属性集合	'getattrs': lambda ins, *args, **kwargs: ins.__class__.get_to_dict_attrs(ins, *args, **kwargs) # 通过类的get_to_dict_attrs获取类实例的所有属性集合
      recursion_value	string	指定循环引用对象的输出替代值 如果不指定则会自动忽略循环引用的对象	'{...}' --如果有循环引用，则用{...}字符串代替
      aa.bb.cc	dict	层级关系的输出选项 请注意级联关系转换配置的键值key采用的是平铺模式 aa.bb.cc表示实例的[aa]关系数据的[bb]关系数据的[cc]关系数据的输出选项 层级关系中也可以设置cascade，如果不设置，子项的cascade=父项的cascade-1	'addresses':{ include:["id","city"] # addresses列表中的Address对象只输出id和city属性 #... }

  • 返回：包含实例属性的字典
  • 示例：

                                   
                                        ins_to_dict(user, {
                                            'cascade': 1,               # 除了user对象本身的属性，addresses关系数据也会输出到字典中
                                            'exclude': ['password'],    # password属性不会输出到字典中
                                            'addresses': {              # addresses关系列表中Address对象的转换选项
                                                'include': ['city']     # 只输出Address对象的city属性
                                            }
                                        })
                                    
                                

发送请求

以下函数用于向其他服务发送API请求

• request(url, method="GET", url_params=None, base_url="", raw_response=False, **kwargs)
  • 功能：发送API请求访问，是requests包的封装
  • 参数：
    • url --请求API的url地址，可以是url字符串或者包含url的字典对象

      -'/api/user/register/'：url字符串

      -{'url':'/api/user/register/', 'method':'POST'}：包含url字符串和method，请注意url对象中包含的method优先级高于method参数

    • method --请求使用的method
    • url_params --url中用到的参数，会将url中的相关的参数变量替换成具体的值
    • base_url --用来生成请求url，适用于对同一个服务器的大量请求场景，可以把url中相同的部分(地址/端口/前缀)放到base_url中，以方便整体修改
    • raw_response

      -True，返回请求对象<Response>res

      -False，返回请求结果(res.text)

    • kwargs --其他请求参数(关键字参数)，包括请求数据，授权信息等，可以参考requests的Session.request方法，常用参数如下。
      • json -请求发送的json数据
      • auth -请求的授权信息，例如，auth=HTTPDigestAuth(username='admin', password='cisco123')
  • 返回：有三种类型的返回值
    1. 请求结果 --请求结果，即res.text，默认返回值
    2. 请求对象<Response>res --raw_response=True时返回，拿到请求对象以后再进行处理
    3. 包含标志位和原因的元组(res_status_codes.api_request_err, result)
  • 示例：

                                   
                                        # 示例1，返回请求对象
                                        def ncs_request(url, **kwargs):
                                            result = request(url, raw_response=True, **ReqConfig, **kwargs)     # 返回请求对象
                                            if type(result) == tuple:                                           # 如果返回结果为元组，则表示请求异常
                                                return False, res_status_codes.api_request_err
    
                                            status_code = result.status_code                                    # 获取请求状态
                                            if status_code // 100 == 2:
                                                return True, result.text                                        # 返回请求结果
                                            return False, res_status_codes.api_request_err
    
    
                                        # 示例2，返回请求结果
                                        json_data = request(current_app.config.get('ELASTICSEARCH_URI'), json=query_json)
                                        json_data = json.loads(json_data)
                                    
                                

• api_request(url, method="GET", url_params=None, base_url="", raw_response=False, **kwargs)

  为了方便使用，flaskz>=1.8.0版本添加flaskz.utils.request函数以替代api_request函数，两者区别如下:

  • 如果请求成功，api_request返回请求结果，request返回(True,请求结果)
  • 如果请求失败，api_request返回(res_status_codes.api_request_err, 失败原因)，request返回(False,失败原因)

  函数描述

  • 功能：发送API请求访问，是requests包的封装
  • 参数：
    • url --请求API的url地址，可以是url字符串或者包含url的字典对象

      -'/api/user/register/'：url字符串

      -{'url':'/api/user/register/', 'method':'POST'}：包含url字符串和method，请注意url对象中包含的method优先级高于method参数

    • method --请求使用的method
    • url_params --url中用到的参数，会将url中的相关的参数变量替换成具体的值
    • base_url --用来生成请求url，适用于对同一个服务器的大量请求场景，可以把url中相同的部分(地址/端口/前缀)放到base_url中，以方便整体修改
    • raw_response

      -True，返回请求对象<Response>res

      -False，返回请求结果(res.text)

    • kwargs --其他请求参数(关键字参数)，包括请求数据，授权信息等，可以参考requests的Session.request方法，常用参数如下。
      • json -请求发送的json数据
      • auth -请求的授权信息，例如，auth=HTTPDigestAuth(username='admin', password='cisco123')
  • 返回：有三种类型的返回值
    1. 请求结果 --请求结果，即res.text，默认返回值
    2. 请求对象<Response>res --raw_response=True时返回，拿到请求对象以后再进行处理
    3. 包含标志位和原因的元组(res_status_codes.api_request_err, result)
  • 示例：

                                   
                                        # 示例1，返回请求对象
                                        def ncs_request(url, **kwargs):
                                            result = api_request(url, raw_response=True, **ReqConfig, **kwargs)     # 返回请求对象
                                            if type(result) == tuple:                                           # 如果返回结果为元组，则表示请求异常
                                                return False, res_status_codes.api_request_err
    
                                            status_code = result.status_code                                    # 获取请求状态
                                            if status_code // 100 == 2:
                                                return True, result.text                                        # 返回请求结果
                                            return False, res_status_codes.api_request_err
    
    
                                        # 示例2，返回请求结果
                                        json_data = api_request(current_app.config.get('ELASTICSEARCH_URI'), json=query_json)
                                        json_data = json.loads(json_data)
                                    
                                

• forward_request(url, payload=None, raw_response=False, error_code=500, **kwargs)
  • 功能：将收到的请求发送到其他url，并将结果返回，内部是通过调用api_request函数进行的转发，一般用于不同系统间通过API进行对接调用的场景
  • 参数：
    • url --转发要请求的url，参考api_request的url参数说明
    • payload --需要转发的payload属性列表，默认转发的属性列表为['method', 'data', 'json', 'headers', 'cookies']

      程序会对payload列表中的值进行遍历，然后将request请求上下文中的相关数据转发出去

    • raw_response --是否返回请求对象，参考api_request的raw_response参数说明
    • error_code --发生异常时要返回的状态码，默认为500
    • kwargs --请求的其他参数，请参考api_request函数
  • 返回：有三种类型的返回值
    1. 请求结果&状态码&header信息组成的元组，(res.text, res.status_code, res.headers.items())，默认返回值
    2. 请求对象<Response>res --raw_response=True时返回
    3. 发生异常时，返回异常原因和error_code组成的元组，(reason, error_code)
  • 示例：

                                   
                                        # 将收到的/auth/tokens/请求转发到其他服务器
                                        @forward_api_bp.route('/auth/tokens/', methods=['POST'])
                                        def remote_auth_token():
                                            return forward_request({'url': 'http://10.124.4.235:8080/api/v1.1/auth/tokens', 'method': 'POST'})
    
                                        # 通过正则表达式转发所有请求
                                        HTTP_METHODS = ['GET', 'HEAD', 'POST', 'PUT', 'DELETE', 'CONNECT', 'OPTIONS', 'TRACE', 'PATCH']
                                        @api_bp.route('/<regex(".*"):path>/', methods=HTTP_METHODS)
                                        def remote(path):
                                            remote_res = forward_request(base_url + path)
                                            res = make_response(remote_res[0], remote_res[1])
                                            # for k, v in remote_res[2]:
                                            #     if k not in ['Transfer-Encoding']:
                                            #         res.headers[k] = v
                                            return res
                                    
                                

• append_url_search_params(url, params)
  • 功能：生成带查询参数的url
  • 参数：
    • url --要附加参数的url
    • params --要附加的查询参数
  • 返回：附加了查询参数的url
  • 示例：

                                   
                                        append_url_search_params('https://example.com',{'foo':1,'bar':2, 'xx': None}) # 'https://example.com?foo=1&bar=2&xx' # append
                                        append_url_search_params('https://example.com?foo=1&bar=2',{'baz':3}) # 'https://example.com?foo=1&bar=2&baz=3' # append
                                        append_url_search_params('https://example.com?foo=1&bar=2',{'bar':3}) # 'https://example.com?foo=1&bar=3' # replace
                                        append_url_search_params('a/b',{'c':3,'d':None}) # 'a/b?c=3&d'
    
                                        append_url_search_params('https://example.com', ['a', 'b', 'c=2']) # 'https://example.com?a&b&c=2'
                                        append_url_search_params('https://example.com', 'a=1&b=2&c')  # 'https://example.com?a=1&b=2&c'
                                    
                                

请求数据处理

以下数据用于请求数据和信息的获取

• get_remote_addr()
  • 功能：获取当前请求的IP地址信息，可用于日志记录等
  • 返回：发送请求的客户端IP地址信息
  • 示例：

                                   
                                        def _get_user_info():   # 获取当前登录用户的相关信息
                                            if current_user.is_anonymous:
                                                return {
                                                    'user_ip': get_remote_addr(),
                                                }
                                            return {
                                                'username': current_user.username,
                                                'user_name': current_user.name,
                                                'user_ip': get_remote_addr()
                                            }
                                    
                                

• is_ajax()
  • 功能：判断当前的请求是否是ajax调用，是通过header中的X-Requested-With是否等于'XMLHttpRequest'字符串进行的判断
  • 返回：

    -True，如果是ajax请求

    -False，如果不是ajax请求

  • 示例：

                                   
                                        def _return_error(app_code, http_code):     # 应用异常处理
                                            app_msg = str(get_status_msg(app_code))
                                            flaskz_logger.error(get_rest_log_msg(request.url + ' request ' + app_msg, None, False, app_code))
    
                                            if is_ajax() or (request.accept_mimetypes.accept_json and not request.accept_mimetypes.accept_html):    #如果是ajax请求，则返回状态码
                                                return create_response(False, app_code)
                                            return get_status_msg(app_msg)  # 如果是其他请求，比如页面html请求，则返回异常信息
                                    
                                

• get_pss(cls, pss_config=None)

  将对模型数据的分页，排序，搜索等查询参数进行解析并生成对应的方法和参数，请参考 flaskz.models.parse_pss

请求响应

以下函数用于生成请求处理结果信息

• create_response(success, data, wrapper=False)
  • 功能：将请求结果数据进行封装，添加status/message等信息，封装以后的数据用于返回到请求客户端，

    一般所有的路由处理函数都会调用create_response生成响应结果数据

  • 参数：
    • success --请求处理是否成功
    • data -返回到请求端的数据，可能是成功数据，也可能是失败原因
    • data_wrapped -返回的请求数据是否已经进行了封装，默认为False(没有进行封装)，逻辑如下

                                             
                                                  if data_wrapped is True:    # 如果已经进行了封装，
                                                      _data = {
                                                          'status': status,       # 附加status信息到封装了以后的数据上
                                                      }
                                                      _data.update(data)
                                                      return _data
                                                  else:                       # 如果没有进行封装
                                                      return {
                                                          'status': status,       # 将status和data封装成一个对象并返回
                                                          'data': data
                                                      }
                                              
                                          

  • 返回：可用于返回到客户端的封装数据
    • 处理成功，返回数据格式如下

                                             
                                                  # 客户端获取到的成功数据示例
                                                  {
                                                    "status": "success",      # 成功status标志，可以通过FLASKZ_RES_SUCCESS_STATUS配置，默认为"success"
                                                    "data": {                 # 成功返回的数据
                                                      "token": "eyJhbGciO......"
                                                    }
                                                  }
                                               
                                          

    • 处理失败，返回数据格式如下，

      请注意为了方便客户端使用，此处返回的status_code是系统定义好的状态码(res_status_codes.db_add_err/res_status_codes.db_data_not_found/...)， 而不是像200/400/500一样的网络请求状态码

                                             
                                                  {
                                                    "status": "fail",                     # 失败status标志，可以通过FLASKZ_RES_FAIL_STATUS配置，默认为"fail"
                                                    "message": "Wrong Password",          # 失败消息信息
                                                    "status_code": "account_verify_err"   # 失败状态代码，用于客户端的逻辑处理，
                                                                                          # 例如，如果客户端接收到status="fail"而且status_code="uri_unauthorized"的消息，就会提示用户登录
                                                  }
                                               
                                          

  • 示例：

                                   
                                        @sys_mgmt_bp.route('/auth/token/', methods=['POST'])    # 用户登录token获取
                                        def sys_auth_get_token():
                                            request_json = request.json
                                            result = User.verify_password(request_json.get('username'), request_json.get('password'))   # 账号/密码校验
                                            success = result[0]         # 处理结果
                                            if success is False:
                                                res_data = model_to_dict(result[1])     # 校验失败，则返回失败原因
                                            else:
                                                res_data = {'token': generate_token({'id': result[1].get_id()})}    # 校验成功，返回token
    
                                            flaskz_logger.info(get_rest_log_msg('User get login token', {'username': request_json.get('username')}, success, res_data))
                                            return create_response(success, res_data)   # 创建返回结果，并返回
                                     
                                

• get_status_msg(status_code)
  • 功能：根据状态码，返回message消息文本，
  • 参数：
    • status_code --状态码，默认提供的状态码定义于res_status_codes模块中
  • 返回：状态码对应的message消息
  • 示例：

                                   
                                        def _return_error(app_code, http_code): # 应用全局异常处理
                                            app_msg = str(get_status_msg(app_code))     # 获取到状态码对应的异常消息
                                            flaskz_logger.error(get_rest_log_msg(request.url + ' request ' + app_msg, None, False, app_code))
    
                                            if is_ajax() or (request.accept_mimetypes.accept_json and not request.accept_mimetypes.accept_html):
                                                return create_response(False, app_code)
                                            # return (app_code, http_code)
                                            return get_status_msg(app_msg)  # 将异常消息发送到客户端
                                    
                                

字典操作

以下函数对于字典操作进行了扩展

• get_deep(d, key, key_split='.', default=None, raising=False)
  • 功能：根据层次键路径，逐层获取对应的键值，目前只有对属性值是dict的层次处理，如果属性值是list，则不会进行处理
  • 参数：
    • d --要获取的字典对象
    • key -层级型键路径，一般是用.连接的字符串，例如,"address.city"，代表的就是获取字典的address对象(字典)的city值
    • key_split -层级型键路径的拆分字符串，通过key_split将字符串拆分成层级数据，默认是.
    • default -如果层级数据不存在，返回的默认值，类似于字典get方法的default用法，默认是None
    • raising -如果层级数据不存在，是否引发异常，默认不会引发异常
  • 返回：层级键路径对应的属性值，如果路径对应的属性值不存在，则返回默认值，或引发异常
  • 示例：

                                   
                                        user = {
                                            "name": "taozh",
                                            "age": 18,
                                            "address": {
                                                "city":"Shanghai",
                                                "district":"Changning"
                                            }
                                       }
                                       print(get_deep(user,"address.city")) # 返回结果为"Shanghai"
                                     
                                

• set_deep(d, key, value, key_split='.')
  • 功能：为字典设置层级键值数据，如果对的层级不存在，则会自动创建并设置属性
  • 参数：
    • d --要设置的字典对象
    • key -层级型键路径，一般是用.连接的字符串，例如,"address.city"，代表的就是字典的address对象(字典)的city属性
    • value -要进行设置的键值数据.
    • key_split -层级型键路径的拆分字符串，通过key_split将字符串拆分成层级数据，默认是.
  • 示例：

                                   
                                    user = {
                                        "name": "taozh",
                                        "age": 18
                                    }
                                    set_deep(user, "address.city", "Shanghai") # 为user的address设置city属性，因为address不存在，所以会先创建一个address字典
                                    # {'name': 'taozh', 'age': 18, 'address': {'city': 'Shanghai'}}
                                     
                                

• get_ins_mapping(ins_list, attr, deep=False)
  • 功能：根据指定属性值对实例对象进行分类，并返回分类映射字典
  • 参数：
    • ins_list --要进行分类的实例对象列表
    • attr -指定的分类属性
    • deep -是否使用层级属性进行分类，如果为True，则会逐层获取到对应的值用于分类
  • 返回：根据指定属性生成的分类映射字典
  • 示例：

                                    
                                        device_id_map = get_ins_mapping(devices, 'model_id') # 根据设备的型号id进行分类，此处的devices是Device类对象列表
                                        device_lb_addr_map = get_ins_mapping(devices, 'ext.loopback_address', True) # 根据设备的ext级联属性的loopback_address地址进行分类
                                    
                                

• get_dict_mapping(dict_list, key='id', key_join="+")
  • 功能：根据一个或多个指定的属性值对字典对象进行分类，并返回分类映射字典
  • 参数：
    • dict_list --要进行分类的字典对象列表
    • key -指定的分类属性，默认通过id属性进行分类，key可以是某个属性也可以是一个由多个属性组成的列表

      如果key是多个属性组成的列表，则会从字典中将多个值取出合并成一个分类键值，类似于联合主键

    • key_join -当key是多个属性组成的列表时，用于合并多个键值并生成分类键值的连接字符
  • 返回：根据指定属性生成的分类映射字典
  • 示例：

                                    
                                        device_id_map = get_dict_map(devices) # 根据设备的id进行分类，此处的devices是包含字典对象的列表
                                        device_name_map = get_dict_map(devices,"name") # 根据设备的name进行分类
                                        policy_device_map = get_dict_map(sr_policies, ['device_id', 'policy_name']) # 根据policy对象的device_id和policy_name进行分类
                                    
                                

• merge_dict(d, *to_merged_dict)
  • 功能：将多个字典的值合并到一个字典中，和字典的update方法主要区别是

    -update方法只会对顶层属性进行合并

    -merge_dict函数会遍历层级属性，并进行合并

  • 参数：
    • d --目标字典，其他字典的属性都会合并到目标字典中
    • *to_merged_dict --一个或多个源字典(位置参数)
  • 示例：

                                    
                                        user = {
                                            "name": "taozh",
                                            "address": {
                                                "city": "Shanghai"
                                            }
                                        }
                                        merge_dict(user,
                                                   {"age": 18},
                                                   {"address": {
                                                       "district": "Changning"  # 会将district属性合并到user的address中
                                                   }})
                                        print(user)
                                        # {'name': 'taozh', 'address': {'city': 'Shanghai', 'district': 'Changning'}, 'age': 18}
    
                                        user1 = {
                                            "name": "taozh",
                                            "address": {
                                                "city": "Shanghai"
                                            }
                                        }
                                        user1.update({
                                            "address": {
                                                "district": "Changning"     # 会将user1的address属性进行整体替换
                                            }
                                        })
                                        print(user1)
                                        # {'name': 'taozh', 'address': {'district': 'Changning'}}
                                    
                                

• del_dict_keys(d, keys)
  • 功能：一次性从字典对象中删除多个键值
  • 参数：
    • d --要进行删除属性的字典对象
    • keys --要删除的键值列表
  • 示例：

                                    
                                        del_dict_keys(json_, ['model', 'vendor']) # 将model和vendor属性从json_中删除
                                    
                                

• pop_dict_keys(d, keys)
  • 功能：一次性从字典对象中pop多个键值
  • 参数：
    • d --要进行pop的字典对象
    • keys --要pop的键值列表
  • 返回：由pop的键值组成的字典
  • 示例：

                                    
                                        pop_dict_keys(a_dict, ['name', 'age'])
                                        pop_dict_keys(a_dict, 'name')
                                     
                                

其他

• filter_list(items, func=None, with_index=False, not_none=False)
  • 功能：过滤指定列表，并返回由符合条件的项组成的列表
  • 参数：
    • items --指定的列表
    • func --过滤函数，返回True表示列表项符合条件
    • with_index --过滤函数是否使用index参数
    • not_none --是否过滤非空列表项
  • 返回：由符合条件的项组成的列表
  • 示例：

                                    
                                        filter_list(items, not_none=True) # 返回非空列表项
                                        filter_list(items, lambda item: item is not None)
                                        filter_list(items, lambda index, item: index > 10 and item is not None, True)
                                     
                                

• find_list(items, func, with_index=False)
  • 功能：从列表中查找第一个符合条件的列表项
  • 参数：
    • items --指定的列表
    • func --查找函数，返回True表示列表项符合条件
    • with_index --查找函数是否使用index参数
  • 返回：第一个符合条件的列表项
  • 示例：

                                    
                                        find_list(items, lambda item: item.get('id') == 10)
                                        find_list(items, lambda index, item: index > 0 and item.get('id') == 10 , True)
                                    
                                

• each_list(items, func, with_index=True)
  • 功能：遍历列表中的列表项，并执行指定的函数
  • 参数：
    • items --指定的列表
    • func --要执行的函数，如果返回False会中断遍历
    • with_index --执行的函数是否使用index参数
  • 返回：第一个符合条件的列表项
  • 示例：

                                   
                                        each_list(items, lambda item: item['selected'] = True)
                                   
                                

• merge_list(target_list, *merged_list)
  • 功能：将多个列表中的对象并到目标列表中
  • 参数：
    • target_list --目标列表，其他列表中的对象都会合并到目标列表中
    • *merged_list --一个或多个源列表(位置参数)
  • 返回：目标列表lst
  • 示例：

                                   
                                        nums = [0, 1]
                                        merge_list(nums, [2, 3], [4, 5]) # nums = [0, 1, 2, 3, 4, 5]
                                     
                                

SSH操作

flaskz.ext.ssh是基于Paramiko封装的SSH操作类(pip install paramiko)，对命令执行和返回结果等进行了处理

• ssh_run_command(connect_kwargs: dict, command: Union[str, list], run_kwargs: Optional[dict] = None) -> Tuple[bool, str]
  • 功能：在指定的host上执行的command
  • 参数：
    • connect_kwargs --主机参数，例如）{'hostname': hostname, 'username': username, 'password': password}
      • hostname --主机地址
      • username --登录账号
      • password --登录密码
      • port --端口(默认为22)
      • recv_endswith --返回信息的结束符号(默认为['# ', '$ ', ': ', '? '])，用于判断recv是否完成
      • recv_start_delay --开始接收数据的延迟(秒)，默认为0.1，一般不需要设置，如果因为网络较差而导致数据接收不全时，可以考虑将delay调大，过大的delay会影响效率
      • pre_commands --在命令正式执行之前预先执行的命令，例如)['terminal length 0', 'term width 512']
      • timeout --ssh建立连接的timeout时间(默认为10s)，如果超出timeout还未建立连接，会抛出异常
      • retries --ssh连接失败的重试次数(默认不重试)
      • secondary_password --二级enable登录密码，用于enable/sudo等操作，如果命令需要enable登录时，如果secondary_password不为None，会自动输入secondary_password
      • channel_kwargs --channel窗口相关参数，例如){'width': 100000, 'height': 100000, 'timeout':2}
    • command --要执行的命令，例如)show running-config
    • run_kwargs --执行命令&结果处理的相关参数，例如){'recv': False, 'prompt': False}
      • recv --是否等待返回结果(默认为True)
      • clean --是否对返回结果进行clean(默认为True)，如果为False，结果可能会包含Welcome info/Last login info/Path promote等信息
      • prompt --跟指定主机交互的prompt(默认为None)，prompt主要用于判断recv是否完成和对返回结果进行clean， 如果为None，会先获取prompt执行对应的命令，如果为False，则表示禁用prompt功能
  • 返回：(success, result)
    • success --如果运行命令成功为True，否则为False
    • result --如果成功返回command运行结果，否则返回异常信息
  • 示例：

                                    
                                        ssh_run_command({'hostname': hostname, 'username': username, 'password': password},'show running-config')
    
                                        ssh_run_command(
                                            # connect_kwargs
                                            {'hostname': 'host', 'username': 'username', 'password': 'password',  # host
                                             'timeout': 10, 'retries': 3,  # connect timeout and retry
                                             'secondary_password': 'enable_pwd',  # enable password
                                             'pre_commands': ['terminal length 0', 'term width 512'],  # pre_commands
                                             'channel_kwargs': {'width': 1000, 'timeout': 2}},  # channel kwargs
                                            # command
                                            'show running-config',
                                            # run_kwargs
                                            {
                                                'clean': False,  # not clean output info
                                                'prompt': False,  # disable prompt
                                            })
    
                                     
                                

• ssh_run_command_list(connect_kwargs: dict, command_list: list, run_kwargs: Optional[dict] = None) -> Tuple[bool, Union[list, str]]
  • 功能：在指定的host上执行多个command
  • 参数：
    • connect_kwargs --主机参数，例如）{'hostname': hostname, 'username': username, 'password': password}
      • hostname --主机地址
      • username --登录账号
      • password --登录密码
      • port --端口(默认为22)
      • recv_endswith --返回信息的结束符号(默认为['# ', '$ ', ': ', '? '])，用于判断recv是否完成
      • recv_start_delay --开始接收数据的延迟(秒)，默认为0.1，一般不需要设置，如果因为网络较差而导致数据接收不全时，可以考虑将delay调大，过大的delay会影响效率
      • pre_commands --在命令正式执行之前预先执行的命令，例如)['terminal length 0', 'term width 512']
      • timeout --ssh建立连接的timeout时间(默认为10s)，如果超出timeout还未建立连接，会抛出异常
      • retries --ssh连接失败的重试次数(默认不重试)
      • secondary_password --二级enable登录密码，用于enable/sudo等操作，如果命令需要enable登录时，如果secondary_password不为None，会自动输入secondary_password
      • channel_kwargs --channel窗口相关参数，例如){'width': 100000, 'height': 100000, 'timeout':2}
    • command_list --要执行的命令列表，例如)['show version', 'show running-config']
    • run_kwargs --执行命令&结果处理的相关参数，例如){'recv': False, 'prompt': False, 'last_result': True}
      • recv --是否等待返回结果(默认为True)
      • clean --是否对返回结果进行clean(默认为True)，如果为False，结果可能会包含Welcome info/Last login info/Path promote等信息
      • prompt --跟指定主机交互的prompt(默认为None)，prompt主要用于判断recv是否完成和对返回结果进行clean， 如果为None，会先获取prompt执行对应的命令，如果为False，则表示禁用prompt功能
      • last_result --是否值返回最后一个command的结果(默认为False)，如果为True，只返回最后一个命令的结果
  • 返回：(success, result_list)
    • success --如果运行命令成功为True，否则为False
    • result_list --如果成功返回command运行结果列表，否则返回异常信息
  • 示例：

                                    
                                        ssh.run_command_list({'hostname': hostname, 'username': username, 'password': password}, ['show version', 'show running-config'])
    
                                        ssh_run_command_list(
                                            # connect_kwargs
                                            {'hostname': 'host', 'username': 'username', 'password': 'password',  # host
                                             'timeout': 10, 'retries': 3,  # connect timeout and retry
                                             'secondary_password': 'enable_pwd',  # enable password
                                             'pre_commands': ['terminal length 0', 'term width 512'],  # pre_commands
                                             'channel_kwargs': {'width': 1000, 'timeout': 2}},  # channel kwargs
                                            # command
                                            ['enable', 'show version'],
                                            # run_kwargs
                                            {
                                                'last_result': True,  # not clean output info
                                                'prompt': False,  # disable prompt
                                            })
                                     
                                

• ssh_session(hostname: str, username: str, password: str = None, port: Optional[int] = 22, **kwargs)
  • 功能：ssh操作上下文管理器
  • 参数：
    • hostname --主机地址
    • username --登录账号
    • password --登录密码
    • port --端口(默认为22)
    • recv_endswith --返回信息的结束符号(默认为['# ', '$ ', ': ', '? '])，用于判断recv是否完成
    • recv_start_delay --开始接收数据的延迟(秒)，默认为0.1，一般不需要设置，如果因为网络较差而导致数据接收不全时，可以考虑将delay调大，过大的delay会影响效率
    • pre_commands --在命令正式执行之前预先执行的命令，例如)['terminal length 0', 'term width 512']
    • timeout --ssh建立连接的timeout时间(默认为10s)，如果超出timeout还未建立连接，会抛出异常
    • secondary_password --二级enable登录密码，用于enable/sudo等操作，如果命令需要enable登录时，如果secondary_password不为None，会自动输入secondary_password
    • channel_kwargs --channel窗口相关参数，例如){'width': 100000, 'height': 100000, 'timeout':2}
  • 返回：flaskz.ext.ssh.SSH对象
  • 示例：

                                    
                                        with ssh_session(host, username, password, timeout=20) as ssh:
                                            ssh.run_command('ls -l')
    
                                        with ssh_session(host, username, password, timeout=20) as ssh:
                                            ssh.run_command_list(['show version', 'show running-config'])
    
                                        with ssh_session(host, username, password, timeout=20, secondary_password=enable_pwd, recv_endswith=['# ', '$ ', ': ', '? ', '#']) as ssh:
                                            ssh.run_command_list(['enable', 'show run'])
                                     
                                

注意事项

• 一次获取全部结果

  有些设备对于命令的返回结果会强制分页(空格/回车继续)，即便命令终端窗口设置的足够大也可能会发生，可以通过执行terminal length 0命令强制不分页

• 数据/结果接收不全

  paramiko执行SSH命令的结果不是一次性返回的，而是类似一个池，ssh一直往里写数据，客户端要从池中循环&分批读取数据，所以判断何时接收完成就很重要

  flaskz.ext.ssh通过对读取的结果跟命令提示符和recv_endswith(['# ', '$ ', ': ', '? '])列表项进行对比来检查接收是否完成

  由于不同的系统和命令返回的结果差异较大，有可能正好匹配到recv_endswith默认列表中的项，从而可能导致数据接收不全的情况

  例如) 执行 `show int | json`，分批获取的命令执行结果时，获取到的某段文本最后可能正好是[冒号+空格](: )，而(: )在recv_endswith列表中，如果此时写入数据稍有延迟，就可能导致程序认为数据已经接收完毕

  可以通过设置recv_endswith列表以避免这种接收不全的情况，例如)设置recv_endswith=['# ', '$ ', ':', '? ']

其他