Flask(flask_sqlalchemy)使用原生sql,多个数据库用法进行封装

新建一个py文件db_seeson.py

<span style="color: #0000ff;">from</span> init <span style="color: #0000ff;">import</span><span style="color: #000000;"> db
</span><span style="color: #0000ff;">def</span> fetch_to_dict(sql, params={}, fecth=<span style="color: #800000;">‘</span><span style="color: #800000;">all</span><span style="color: #800000;">‘</span>, bind=<span style="color: #000000;">None):
    </span><span style="color: #800000;">‘‘‘</span><span style="color: #800000;">
    dict的方式返回数据
    :param sql: select * from xxx where name=:name
    :param params:{‘name‘:‘zhangsan‘}
    :param fecth:默认返回全部数据，返回格式为[{},{}],如果fecth=‘one‘,返回单条数据，格式为dict
    :param bind:连接的数据，默认取配置的SQLALCHEMY_DATABASE_URL，
    :return:
    </span><span style="color: #800000;">‘‘‘</span><span style="color: #000000;">
    resultProxy </span>= db.session.execute(sql, params, bind=db.get_engine(bind=<span style="color: #000000;">bind))
    </span><span style="color: #0000ff;">if</span> fecth == <span style="color: #800000;">‘</span><span style="color: #800000;">one</span><span style="color: #800000;">‘</span><span style="color: #000000;">:
        result_tuple </span>=<span style="color: #000000;"> resultProxy.fetchone()
        </span><span style="color: #0000ff;">if</span><span style="color: #000000;"> result_tuple:
            result </span>=<span style="color: #000000;"> dict(zip(resultProxy.keys(), list(result_tuple)))
        </span><span style="color: #0000ff;">else</span><span style="color: #000000;">:
            </span><span style="color: #0000ff;">return</span><span style="color: #000000;"> None
    </span><span style="color: #0000ff;">else</span><span style="color: #000000;">:
        result_tuple_list </span>=<span style="color: #000000;"> resultProxy.fetchall()
        </span><span style="color: #0000ff;">if</span><span style="color: #000000;"> result_tuple_list:
            result </span>=<span style="color: #000000;"> []
            keys </span>=<span style="color: #000000;"> resultProxy.keys()
            </span><span style="color: #0000ff;">for</span> row <span style="color: #0000ff;">in</span><span style="color: #000000;"> result_tuple_list:
                result_row </span>=<span style="color: #000000;"> dict(zip(keys, row))
                result.append(result_row)
        </span><span style="color: #0000ff;">else</span><span style="color: #000000;">:
            </span><span style="color: #0000ff;">return</span><span style="color: #000000;"> None
    </span><span style="color: #0000ff;">return</span><span style="color: #000000;"> result
</span><span style="color: #008000;">#</span><span style="color: #008000;"> 分页</span>
<span style="color: #0000ff;">def</span> fetch_to_dict_pagetion(sql, params={}, page=1, page_size=15, bind=<span style="color: #000000;">None):
    sql_count </span>= <span style="color: #800000;">"""</span><span style="color: #800000;">select count(*) as count from (%s) _count</span><span style="color: #800000;">"""</span> %<span style="color: #000000;"> sql
    total_count </span>= get_count(sql_count, params, bind=<span style="color: #000000;">bind)
    sql_page </span>= <span style="color: #800000;">‘</span><span style="color: #800000;">%s limit %s,%s</span><span style="color: #800000;">‘</span> % (sql, (page - 1) *<span style="color: #000000;"> page_size, page_size)
    </span><span style="color: #0000ff;">print</span>(<span style="color: #800000;">‘</span><span style="color: #800000;">sql_page:</span><span style="color: #800000;">‘</span><span style="color: #000000;">, sql_page)
    result </span>= fetch_to_dict(sql_page, params, <span style="color: #800000;">‘</span><span style="color: #800000;">all</span><span style="color: #800000;">‘</span>, bind=<span style="color: #000000;">bind)
    result_dict </span>= {<span style="color: #800000;">‘</span><span style="color: #800000;">results</span><span style="color: #800000;">‘</span>: result, <span style="color: #800000;">‘</span><span style="color: #800000;">count</span><span style="color: #800000;">‘</span><span style="color: #000000;">: total_count}
    </span><span style="color: #0000ff;">return</span><span style="color: #000000;"> result_dict
</span><span style="color: #008000;">#</span><span style="color: #008000;"> 执行单条语句（update,insert）</span>
<span style="color: #0000ff;">def</span> execute(sql, params={}, bind=<span style="color: #000000;">None):
    </span><span style="color: #0000ff;">print</span>(<span style="color: #800000;">‘</span><span style="color: #800000;">sql</span><span style="color: #800000;">‘</span><span style="color: #000000;">, sql)
    db.session.execute(sql, params, bind</span>=db.get_engine(bind=<span style="color: #000000;">bind))
    db.session.commit()
</span><span style="color: #008000;">#</span><span style="color: #008000;"> 执行多条语句，失败自动回滚</span>
<span style="color: #0000ff;">def</span><span style="color: #000000;"> execute_many(sqls):
    </span><span style="color: #0000ff;">print</span><span style="color: #000000;">(sqls)
    </span><span style="color: #0000ff;">if</span> <span style="color: #0000ff;">not</span><span style="color: #000000;"> isinstance(sqls, (list, tuple)):
        </span><span style="color: #0000ff;">raise</span> Exception(<span style="color: #800000;">‘</span><span style="color: #800000;">type of the parameters must be list or tuple</span><span style="color: #800000;">‘</span><span style="color: #000000;">)
    </span><span style="color: #0000ff;">if</span> len(sqls) ==<span style="color: #000000;"> 0:
        </span><span style="color: #0000ff;">raise</span> Exception(<span style="color: #800000;">"</span><span style="color: #800000;">parameters‘s length can‘t be 0</span><span style="color: #800000;">"</span><span style="color: #000000;">)
    </span><span style="color: #0000ff;">for</span> statement <span style="color: #0000ff;">in</span><span style="color: #000000;"> sqls:
        </span><span style="color: #0000ff;">if</span> <span style="color: #0000ff;">not</span><span style="color: #000000;"> isinstance(statement, dict):
            </span><span style="color: #0000ff;">raise</span> Exception(<span style="color: #800000;">"</span><span style="color: #800000;">parameters erro</span><span style="color: #800000;">"</span><span style="color: #000000;">)
    </span><span style="color: #0000ff;">try</span><span style="color: #000000;">:
        </span><span style="color: #0000ff;">for</span> s <span style="color: #0000ff;">in</span><span style="color: #000000;"> sqls:
            db.session.execute(s.get(</span><span style="color: #800000;">‘</span><span style="color: #800000;">sql</span><span style="color: #800000;">‘</span>), s.get(<span style="color: #800000;">‘</span><span style="color: #800000;">params</span><span style="color: #800000;">‘</span>), bind=db.get_engine(bind=s.get(<span style="color: #800000;">‘</span><span style="color: #800000;">bind</span><span style="color: #800000;">‘</span><span style="color: #000000;">, None)))
        db.session.commit()
    </span><span style="color: #0000ff;">except</span><span style="color: #000000;"> Exception as e:
        db.session.rollback()
        </span><span style="color: #0000ff;">raise</span> Exception(<span style="color: #800000;">"</span><span style="color: #800000;">execute sql fail ,is rollback</span><span style="color: #800000;">"</span>)

Flask(flask_sqlalchemy)使用原生sql,多个数据库用法进行封装

标签：rom   statement   int   数据库   自动   append   man   fail   ==