python manage.py runserver

  FIND_BY_SQL = "findBySql" # 根据sql查找
  COUNT_BY_SQL = "countBySql" # 自定义sql 统计影响行数
  INSERT = "insert" # 插入
  UPDATE_BY_ATTR = "updateByAttr" # 更新数据
  DELETE_BY_ATTR = "deleteByAttr" # 删除数据
  FIND_BY_ATTR = "findByAttr" # 根据条件查询一条记录
  FIND_ALL_BY_ATTR = "findAllByAttr"  #根据条件查询多条记录
  COUNT = "count" # 统计行
  EXIST = "exist" # 是否存在该记录


import mysql.connector
import mysql.connector.errors
from common.customConst import Const
class MySQLet:
    """Connection to a MySQL"""
    # def __init__(self,user='',password='',database='',charset=None,port=3306):
    def __init__(self,**kwargs):
        try:
            self._conn = mysql.connector.connect(host=kwargs["host"], user=kwargs["user"], password=kwargs["password"],
                                                 charset=kwargs["charset"], database=kwargs["database"], port=kwargs["port"])
            self.__cursor = None
            print("连接数据库")
            #set charset charset = ('latin1','latin1_general_ci')
        except mysql.connector.errors.ProgrammingError as err:
            print('mysql连接错误：' + err.msg)

    # def findBySql(self, sql, params={}, limit=0, join='AND'):
    def findBySql(self, **kwargs):
        """
        自定义sql语句查找
        limit = 是否需要返回多少行
        params = dict(field=value)
        join = 'AND | OR'
        """
        cursor = self.__getCursor()
        # sql = self.__joinWhere(kwargs["sql"], kwargs["params"], kwargs["join"])
        if kwargs.get("join", 0) == 0: kwargs["join"] = "AND"
        sql = self.__joinWhere(**kwargs)
        cursor.execute(sql, tuple(kwargs["params"].values()))
        rows = cursor.fetchmany(size=kwargs["limit"]) if kwargs["limit"] > 0 else cursor.fetchall()
        result = [dict(zip(cursor.column_names,row)) for row in rows] if rows else None
        return len(result)

    # def countBySql(self,sql,params = {},join = 'AND'):
    def countBySql(self, **kwargs):
        """自定义sql 统计影响行数"""
        if kwargs.get("join", 0) == 0: kwargs["join"] = "AND"
        cursor = self.__getCursor()
        # sql = self.__joinWhere(kwargs["sql"], kwargs["params"], kwargs["join"])
        sql = self.__joinWhere(**kwargs)
        cursor.execute(sql, tuple(kwargs["params"].values()))
        result = cursor.fetchall() # fetchone是一条记录， fetchall 所有记录
        return len(result) if result else 0

    # def insert(self,table,data):
    def insert(self, **kwargs):
        """新增一条记录
          table: 表名
          data: dict 插入的数据
        """
        fields = ','.join('`'+k+'`' for k in kwargs["data"].keys())
        values = ','.join(("%s", ) * len(kwargs["data"]))
        sql = 'INSERT INTO `%s` (%s) VALUES (%s)' % (kwargs["table"], fields, values)
        cursor = self.__getCursor()
        cursor.execute(sql, tuple(kwargs["data"].values()))
        insert_id = cursor.lastrowid
        self._conn.commit()
        return insert_id

    # def updateByAttr(self,table,data,params={},join='AND'):
    def updateByAttr(self, **kwargs):
    #     """更新数据"""
        if kwargs.get("params", 0) == 0:
            kwargs["params"] = {}
        if kwargs.get("join", 0) == 0:
            kwargs["join"] = "AND"
        fields = ','.join('`' + k + '`=%s' for k in kwargs["data"].keys())
        values = list(kwargs["data"].values())


        values.extend(list(kwargs["params"].values()))
        sql = "UPDATE `%s` SET %s " % (kwargs["table"], fields)
        kwargs["sql"] = sql
        sql = self.__joinWhere(**kwargs)
        cursor = self.__getCursor()
        cursor.execute(sql, tuple(values))
        self._conn.commit()
        return cursor.rowcount


    # def updateByPk(self,table,data,id,pk='id'):
    def updateByPk(self, **kwargs):
        """根据主键更新，默认是id为主键"""
        return self.updateByAttr(**kwargs)

    # def deleteByAttr(self,table,params={},join='AND'):
    def deleteByAttr(self, **kwargs):
        """删除数据"""
        if kwargs.get("params", 0) == 0:
            kwargs["params"] = {}
        if kwargs.get("join", 0) == 0:
            kwargs["join"] = "AND"
        # fields = ','.join('`'+k+'`=%s' for k in kwargs["params"].keys())
        sql = "DELETE FROM `%s` " % kwargs["table"]
        kwargs["sql"] = sql
        # sql = self.__joinWhere(sql, kwargs["params"], kwargs["join"])
        sql = self.__joinWhere(**kwargs)
        cursor = self.__getCursor()
        cursor.execute(sql, tuple(kwargs["params"].values()))
        self._conn.commit()
        return cursor.rowcount

    # def deleteByPk(self,table,id,pk='id'):
    def deleteByPk(self, **kwargs):
        """根据主键删除，默认是id为主键"""
        return self.deleteByAttr(**kwargs)

    # def findByAttr(self,table,criteria = {}):
    def findByAttr(self, **kwargs):
        """根據條件查找一條記錄"""
        return self.__query(**kwargs)

    # def findByPk(self,table,id,pk='id'):
    def findByPk(self, **kwargs):
        return self.findByAttr(**kwargs)

    # def findAllByAttr(self,table,criteria={}, whole=true):
    def findAllByAttr(self, **kwargs):
        """根據條件查找記錄"""
        return self.__query(**kwargs)

    # def count(self,table,params={},join='AND'):
    def count(self, **kwargs):
        """根据条件统计行数"""
        if kwargs.get("join", 0) == 0: kwargs["join"] = "AND"
        sql = 'SELECT COUNT(*) FROM `%s`' % kwargs["table"]
        # sql = self.__joinWhere(sql, kwargs["params"], kwargs["join"])
        kwargs["sql"] = sql
        sql = self.__joinWhere(**kwargs)
        cursor = self.__getCursor()
        cursor.execute(sql, tuple(kwargs["params"].values()))
        result = cursor.fetchone()
        return result[0] if result else 0

    # def exist(self,table,params={},join='AND'):
    def exist(self, **kwargs):
        """判断是否存在"""
        return self.count(**kwargs) > 0

    def close(self):
        """关闭游标和数据库连接"""
        if self.__cursor is not None:
            self.__cursor.close()
        self._conn.close()

    def __getCursor(self):
        """获取游标"""
        if self.__cursor is None:
            self.__cursor = self._conn.cursor()
        return self.__cursor

    # def __joinWhere(self,sql,params,join):
    def __joinWhere(self, **kwargs):
        """转换params为where连接语句"""
        if kwargs["params"]:
            keys,_keys = self.__tParams(**kwargs)
            where = ' AND '.join(k+'='+_k for k,_k in zip(keys,_keys)) if kwargs["join"] == 'AND' else ' OR '.join(k+'='+_k for k,_k in zip(keys,_keys))
            kwargs["sql"]+=' WHERE ' + where
        return kwargs["sql"]

    # def __tParams(self,params):
    def __tParams(self, **kwargs):
        keys = ['`'+k+'`' for k in kwargs["params"].keys()]
        _keys = ['%s' for k in kwargs["params"].keys()]
        return keys,_keys

    # def __query(self,table,criteria,whole=False):
    def __query(self, **kwargs):
        if kwargs.get("whole", False) == False or kwargs["whole"] is not True:
            kwargs["whole"] = False
            kwargs["criteria"]['limit'] = 1
        # sql = self.__contact_sql(kwargs["table"], kwargs["criteria"])
        sql = self.__contact_sql(**kwargs)
        cursor = self.__getCursor()
        cursor.execute(sql)
        rows = cursor.fetchall() if kwargs["whole"] else cursor.fetchone()
        result = [dict(zip(cursor.column_names, row)) for row in rows] if kwargs["whole"] else dict(zip(cursor.column_names, rows)) if rows else None
        return result

    # def __contact_sql(self,table,criteria):
    def __contact_sql(self, **kwargs):
        sql = 'SELECT '
        if kwargs["criteria"] and type(kwargs["criteria"]) is dict:
            #select fields
            if 'select' in kwargs["criteria"]:
                fields = kwargs["criteria"]['select'].split(',')
                sql+= ','.join('`'+field+'`' for field in fields)
            else:
                sql+=' * '
            #table
            sql+=' FROM `%s`'% kwargs["table"]
            #where
            if 'where' in kwargs["criteria"]:
                sql+=' WHERE '+ kwargs["criteria"]['where']
            #group by
            if 'group' in kwargs["criteria"]:
                sql+=' GROUP BY '+ kwargs["criteria"]['group']
            #having
            if 'having' in kwargs["criteria"]:
                sql+=' HAVING '+ kwargs["criteria"]['having']
            #order by
            if 'order' in kwargs["criteria"]:
                sql+=' ORDER BY '+ kwargs["criteria"]['order']
            #limit
            if 'limit' in kwargs["criteria"]:
                sql+=' LIMIT '+ str(kwargs["criteria"]['limit'])
            #offset
            if 'offset' in kwargs["criteria"]:
                sql+=' OFFSET '+ str(kwargs["criteria"]['offset'])
        else:
            sql+=' * FROM `%s`'% kwargs["table"]
        return sql
    def findKeySql(self, key ,**kwargs):
        sqlOperate = {
        Const.COUNT: lambda: self.count(**kwargs),
        Const.COUNT_BY_SQL: lambda: self.countBySql(**kwargs),
        Const.DELETE_BY_ATTR: lambda: self.deleteByAttr(**kwargs),
        Const.EXIST: lambda: self.exist(**kwargs),
        Const.FIND_ALL_BY_ATTR: lambda: self.findAllByAttr(**kwargs),
        Const.INSERT: lambda: self.insert(**kwargs),
        Const.FIND_BY_ATTR: lambda: self.findByAttr(**kwargs),
        Const.UPDATE_BY_ATTR: lambda: self.updateByAttr(**kwargs),
        Const.FIND_BY_SQL: lambda: self.findBySql(**kwargs)

        }
        return sqlOperate[key]()


if __name__ == "__main__":
    mysqlet = MySQLet(host="127.0.0.1", user="root", password="", charset="utf8", database="userinfo", port=3306)
    # 根据字段统计count, join>>AND,OR,可以不传，默认为AND
    print(mysqlet.findKeySql(Const.COUNT, table="info", params={"id": "11", "name": "666"}, join="OR"))
    # 自定义sql语句统计count
    print(mysqlet.findKeySql(Const.COUNT_BY_SQL, sql="select * from info", params={"name": "666"}, join="AND"))
    #插入数据
    print(mysqlet.findKeySql(Const.INSERT, table="info", data={"name":"333", "pwd": "111"}))
    #根据字段删除,不传params参数，就是删除全部
    print(mysqlet.findKeySql(Const.DELETE_BY_ATTR, table="info", params={"id": 20}))
    # 查找是否存在该记录,不传params参数，就是查找全部.join同上
    print(mysqlet.findKeySql(Const.EXIST, table="info", params={"id": 180},join='AND'))
    #根据字段查找多条记录，whole不传就查一条记录，criteria里面可以传where,group by,having,order by,limt,offset
    print(mysqlet.findKeySql(Const.FIND_ALL_BY_ATTR, table="info", criteria= {"where": "name=333"}, whole=True))
    # 根据字段查一条记录，和上面的查多条记录参数基本一样，少了个whole参数
    print(mysqlet.findKeySql(Const.FIND_BY_ATTR, table="info", criteria= {"where": "name=333"}))
    # 根据字段更新数据库中的记录，join可以传AND,OR,不传默认取AND
    print(mysqlet.findKeySql(Const.UPDATE_BY_ATTR, table="info",data={"name": "-09"}, params={"id": 18, "name": "333"}, join='AND'))
    # 根据自定义sql语句查询记录，limit:0表示所有记录，join：AND|OR.不传取AND
    print(mysqlet.findKeySql(Const.FIND_BY_SQL, sql="select * from info", params={"name": "333", "id": 18}, limit=0))

class MyClass(object):
    def __init__(self):
        self.ids = "001"

    def t1(self, path=""):
        print("这个是实例化方法")

    @staticmethod
    def get_result(path):
        print("这个是静态方法")
        MyClass().t1()  # 显示调用类的实例化方法
        print(MyClass().ids)

    @classmethod
    def get_test1(cls, path):
        print("这个是类方法")
        cls().t1()  # 调用类的实例化方法
        print(cls().ids)


MyClass().t1("调用实例化方法")
MyClass.get_result("调用静态方法")
MyClass.get_test1("调用类方法")

class Count():
    def add(self):
        pass

from unittest import mock
import unittest
from modular import Count
class TestCount(unittest.TestCase):

    def test_add(self):
        count = Count()
        count.add = mock.Mock(return_value=13)
        result = count.add(8,5)
        self.assertEqual(result,13)


    if __name__ == '__main__':
        unittest.main() 

pip install HTTPretty

import requests
import httpretty

def test_one():
    httpretty.enable()  # enable HTTPretty so that it will monkey patch the socket module
    httpretty.register_uri(httpretty.GET, "http://yipit.com/",
                           body="Find the best daily deals")

    response = requests.get('http://yipit.com')

    assert response.text == "Find the best daily deals"

    httpretty.disable()  # disable afterwards, so that you will have no problems in code that uses that socket module
    httpretty.reset()    # reset H

import requests
from sure import expect
import httpretty


@httpretty.activate
def test_yipit_api_integration():
    httpretty.register_uri(httpretty.POST, "http://api.yipit.com/foo/",
                           body='{"repositories": ["HTTPretty", "lettuce"]}')

    response = requests.post('http://api.yipit.com/foo',
                            '{"username": "gabrielfalcao"}',
                            headers={
                                'content-type': 'text/json',
                            })

    expect(response.text).to.equal('{"repositories": ["HTTPretty", "lettuce"]}')
    expect(httpretty.last_request().method).to.equal("POST")
    expect(httpretty.last_request().headers['content-type']).to.equal('text/json')

from httmock import urlmatch, HTTMock
import requests

@urlmatch(netloc=r'(.*\.)?google\.com$')
def google_mock(url, request):
    return 'Feeling lucky, punk?'

with HTTMock(google_mock):
    r = requests.get('http://google.com/')
print r.content  # 'Feeling lucky, punk?'

from httmock import all_requests, HTTMock
import requests

@all_requests
def response_content(url, request):
    return {'status_code': 200,
            'content': 'Oh hai'}

with HTTMock(response_content):
    r = requests.get('https://foo_bar')

print r.status_code
print r.content

from httmock import all_requests, response, HTTMock
import requests

@all_requests
def response_content(url, request):
    headers = {'content-type': 'application/json',
               'Set-Cookie': 'foo=bar;'}
    content = {'message': 'API rate limit exceeded'}
    return response(403, content, headers, None, 5, request)

with HTTMock(response_content):
    r = requests.get('https://api.github.com/users/whatever')

print r.json().get('message')
print r.cookies['foo']

map(lambda x: x ** 2, [1, 2, 3, 4, 5])
返回结果为：
[1, 4, 9, 16, 25]

def add100(x):
     return x+100

 hh = [11,22,33]
 map(add100,hh)
 [111, 122, 133]

def is_even(x):
      return x & 1 != 0
filter(is_even, [1, 2, 3, 4, 5, 6, 7, 8, 9, 10])
返回结果为：
[1, 3, 5, 7, 9]

def myadd(x,y):  
        return x+y  
    sum=reduce(myadd,(1,2,3,4,5,6,7))  
    print(sum)  
    #结果就是输出1+2+3+4+5+6+7的结果即28

reduce(lambda x, y: x + y, [2, 3, 4, 5, 6], 1)
结果为21(  (((((1+2)+3)+4)+5)+6)  )
reduce(lambda x, y: x + y, [2, 3, 4, 5, 6])
结果为20

import xlrd  # 导入模块
from xlutils.copy import copy  # 导入copy模块

class EditExcel():

    def __init__(self, path, sheet_name):
        self.path = path
        self.sheet_name = sheet_name

    def get_excel_sheet(self):
        # 打开excel
        read_open_xls = xlrd.open_workbook(self.path)
        # 得到excel中的sheet_name
        read_xlsx_sheet = read_open_xls.sheet_by_name(self.sheet_name)
        # 拷贝excel
        copy_book = copy(read_open_xls)
        return copy_book, read_xlsx_sheet,

    def write_value(self):
        copy_book, read_xlsx_sheet = self.get_excel_sheet()
        # 获取行数
        row_max = read_xlsx_sheet.nrows
        # 获取第一行的值
        rows = read_xlsx_sheet.row_values(0)
        # 获取列数
        col_max = read_xlsx_sheet.ncols
        for row in range(row_max):
            row_value = read_xlsx_sheet.row_values(row)
            for col in range(col_max):
                current_value = read_xlsx_sheet.cell(row, col).value
                # 按照条件过滤
                if current_value == 'Pass':
                    copy_sheet = copy_book.get_sheet(0)
                    copy_sheet.write(row, col, '11111')
                    copy_book.save(self.path)


EditExcel("test.xls", "Sheet1").write_value()

from email.header import Header
from email.mime.text import MIMEText
from email.utils import parseaddr, formataddr
from email.mime.multipart import MIMEMultipart
from email.mime.application import MIMEApplication
import smtplib
# 第三方 SMTP 服务
mail_host="smtp.qq.com"  #设置服务器
mail_user="111@qq.com"    #用户名
mail_pass="adfafda"   #口令

def _format_addr(s):
    name, addr = parseaddr(s)
    return formataddr((Header(name, 'utf-8').encode(), addr))


def send_mail(f, to_addr):
    '''

    :param f: 附件路径
    :param to_addr:发给的人 []
    :return:
    '''
    from_addr = mail_user
    password = mail_pass
    # to_addr = "ashikun@126.com"
    smtp_server = mail_host

    msg = MIMEMultipart()

    # msg = MIMEText('hello, send by Python...', 'plain', 'utf-8')
    msg['From'] = _format_addr('坤少发的邮件<%s>' % from_addr)
    msg['To'] = _format_addr('大人 <%s>' % to_addr)
    msg['Subject'] = Header('接口测试报告……', 'utf-8').encode()

    msg.attach(MIMEText('接口测试报告.', 'plain', 'utf-8'))
    part = MIMEApplication(open(f, 'rb').read())
    part.add_header('Content-Disposition', 'attachment', filename=f)
    msg.attach(part)

    server = smtplib.SMTP_SSL(smtp_server, 465)
    server.set_debuglevel(1)
    server.login(from_addr, password)
    server.sendmail(from_addr, to_addr, msg.as_string())
    server.quit()

import itchat

#产生二维码
itchat.auto_login(hotReload=True)
#定义用户的昵称
send_userid='亲爱的'
#查找用户的userid
itcaht_user_name = itchat.search_friends(name=send_userid)[0]['UserName']
#利用send_msg发送消息
itchat.send_msg('这是一个测试',toUserName=itcaht_user_name)

import requests
import itchat
# 去图灵机器人官网注册后会生成一个apikey，可在个人中心查看
KEY = '8edce3ce905a4c1dbb96**************'
def get_response(msg):
    apiUrl = 'http://www.tuling123.com/openapi/api'
    data = {
        'key'   : KEY,
        'info'   : msg,   # 这是要发送出去的信息
        'userid'  : 'wechat-rebot',  #这里随意写点什么都行
    }
    try:
        # 发送一个post请求
        r = requests.post(apiUrl, data =data).json()
        # 获取文本信息，若没有‘Text’ 值，将返回Nonoe 
        return r.get('text')
    except:
        return
# 通过定义装饰器加强函数 tuling_reply(msg) 功能，获取注册文本信息
@itchat.msg_register(itchat.content.TEXT)
def tuling_reply(msg):
    # 设置一个默认回复，在出现问题仍能正常回复信息
    defaultReply = 'I received: ' +msg['Text']
    reply = get_response(msg['Text'])
    # a or b 表示，如有a有内容，那么返回a，否则返回b
    return reply or defaultReply
# 使用热启动，不需要多次扫码
itchat.auto_login(hotReload=True)
itchat.run()