连接数据库

概述

lessweb内置sqlalchemy管理和操作数据库。

注意:lessweb默认没有依赖sqlalchemy,需要自己手动安装。

连接数据库的方法

使用lessweb.plugin.database的init方法,可以创建一个全局的连接池。

@overload
def init(*, dburi, echo=True, autoflush=True, autocommit=False): ...
@overload
def init(*, protocol, username, password, host, port:int, database, echo=True, autoflush=True, autocommit=False): ...

获取请求级session

使用lessweb.plugin.database的processor拦截器即可,请求级session就会保存在上下文对象的db成员中。

原理

class DatabaseCtx(Context):
    db : Session

def processor(ctx: DatabaseCtx):
    try:
        ctx.db = global_data.db_session_maker()
        if global_data.autocommit:
            with ctx.db.begin():
                return ctx()
        else:
            return ctx()
    except:
        ctx.db.rollback()
        raise
    finally:
        ctx.db.close()

示例

以mysql-connector连接mysql为例:

requirements.txt

sqlalchemy
mysql-connector

controller.py

from lessweb.plugin.database import DbServ

def book_detail(serv:DbServ, id:int):
    ret = serv.db.execute('SELECT * FROM TblBook WHERE id=:id', {'id': id}).first()
    return str(ret)

index.py

from lessweb import Application
from lessweb.plugin import database
from controller import book_detail

database.init(
  protocol='mysql+mysqlconnector',
  username='root', password='pwd',
  host='localhost', port=3306, database='proj', echo=True,
  autoflush=True, autocommit=False
)

app = Application()
app.add_interceptor('.*', method='*', dealer=database.processor)
app.add_get_mapping('/book/{id}', book_detail)
app.run()