Flask sqlalchemy 映射关联已存在的表
1、安装库pip install sqlalchemypip install mysql-connector-python驱动包如果没有安装成功会报错,无法链接!2、示例代码from sqlalchemy import create_engine,Table,MetaDatafrom sqlalchemy.orm import Sessionmetadata = MetaData()engine
·
方法1:sqlalchemy
安装库
pip install sqlalchemy
pip install mysql-connector-python
驱动包如果没有安装成功会报错,无法链接!
示例代码
from sqlalchemy import create_engine,Table,MetaData
from sqlalchemy.orm import Session
metadata = MetaData()
engine = create_engine(
'mysql+mysqlconnector://root:password@localhost:3306/my_db',
echo = True
)
# 获取指定数据库链接对象
Admin = Table('servers',metadata,autoload=True,autoload_with=engine)
session = Session(engine)
# 返回查询数据的第一条记录
res = session.query(Admin).first()
print(res)
方法2:flask_sqlalchemy
示例代码
from flask_sqlalchemy import SQLAlchemy,make_url
def get_mysql_conn_url(config):
"""
:description: 生成sqlalchemy使用的连接url
:param hy_config: hy_config
:return: url
"""
mysql_conn_map = dict(
dialect="mysql",
driver="mysqlconnector",
host=config["MYSQL_HOST"],
port=config["MYSQL_PORT"],
database=config["MYSQL_DB"],
user=config["MYSQL_USER"],
password=config["MYSQL_PASSWORD"],
)
s = "mysql+mysqlconnector://{user}:{password}@{host}:{port}/{database}".format(**mysql_conn_map)
url = make_url(s)
return url
def create_mysql_ORM(app):
"""
创建MySQL的ORM对象并反射数据库中已存在的表,获取所有存在的表对象
:param app: app:flask实例
:return: (db:orm-obj, all_table:数据库中所有已存在的表的对象(dict))
"""
# 创建mysql连接对象
url = get_mysql_conn_url(config=app.config)
app.config["SQLALCHEMY_DATABASE_URI"] = url
app.config["SQLALCHEMY_TRACK_MODIFICATIONS"] = True # 每次请求结束时自动commit数据库修改
app.config["SQLALCHEMY_ECHO"] = False # 如果设置成 True,SQLAlchemy将会记录所有发到标准输出(stderr)的语句,这对调试很有帮助.
app.config["SQLALCHEMY_RECORD_QUERIES"] = None # 可以用于显式地禁用或者启用查询记录。查询记录 在调试或者测试模式下自动启用。
app.config["SQLALCHEMY_POOL_SIZE"] = 5 # 数据库连接池的大小。默认是数据库引擎的默认值(通常是 5)。
app.config["SQLALCHEMY_POOL_TIMEOUT"] = 10 # 指定数据库连接池的超时时间。默认是 10。
"""
自动回收连接的秒数。
这对 MySQL 是必须的,默认 情况下 MySQL 会自动移除闲置 8 小时或者以上的连接。
需要注意地是如果使用 MySQL 的话, Flask-SQLAlchemy 会自动地设置这个值为 2 小时。
"""
app.config["SQLALCHEMY_POOL_RECYCLE"] = None
"""
控制在连接池达到最大值后可以创建的连接数。
当这些额外的 连接回收到连接池后将会被断开和抛弃。
"""
app.config["SQLALCHEMY_MAX_OVERFLOW"] = None
# 获取SQLAlchemy实例对象
db = SQLAlchemy(app)
# 反射数据库中已存在的表,并获取所有存在的表对象。
db.reflect()
all_table = {table_obj.name: table_obj for table_obj in db.get_tables_for_bind()}
return db, all_table
实例化,载入config配置信息
app = Flask(__name__)
app.config.from_object('config.BaseConfig')
db, all_table = create_mysql_ORM(app=app)
查询数据
@app.route('/query',methods=['POST'])
def query():
form_dict = request.form
name = form_dict['name']
limit = form_dict['limit']
page = form_dict['page']
data = {"code": 0,"msg": "Successful",}
try:
table = all_table[name]
_count = db.session.query(db.func.count(table.c.host)).scalar()
# result = db.session.query(table).all()
result = db.session.query(table).filter(table.c.address.like('%上海%')).limit(limit).offset((int(page) - 1))
data['count'] = _count
data['data'] = [x._asdict() for x in result]
except Exception as err:
data['code'] = 1
data['msg'] = 'Failed! {}'.format(err)
return jsonify(data)
返回结果
更多推荐
所有评论(0)