python使用flask框架ORM操作mysql oracle
本文章记录着python使用flask ORM 连接mysql oracle 数据库 的方法
示例一：python调用flask框架
import os
import sys
from flask import Flask
from flask_script import Command, Manager
app = Flask(__name__)
class Print(Command):
"""
只是测试输出的内容
"""
def run(self):
print("hello")
if __name__ == "__main__":
"""
使用方法:
只是输出: python demo.py print
"""
manager = Manager(app)
manager.add_command("print", Print())
manager.run()
调用方法
python demo.py print
输出
hello
示例二：python调用flask连接MySQL
import os
import sys
# sys.path.append(os.path.realpath("../"))
# sys.path.append("/Users/mac/Documents/www_python/wannet/applications")
from flask import Flask
from flask_sqlalchemy import SQLAlchemy
from flask_script import Command, Manager
app = Flask(__name__)
# 配置数据库连接方法一: 单库
# MySQL 连接
app.config['SQLALCHEMY_DATABASE_URI'] = 'mysql+pymysql://root:root@localhost/wannet'
# Oracle 连接
# app.config['SQLALCHEMY_DATABASE_URI'] = "oracle+cx_oracle://dev:dev@localhost:1521/helowin"
# 额外配置
app.config['SQLALCHEMY_COMMIT_ON_TEARDOWN'] = True
app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = False
db = SQLAlchemy(app)
class Print(Command):
"""
只是测试输出的内容
"""
def run(self):
print("hello")
class User(db.Model):
"""
MySQL Model
"""
__tablename__ = 'user'
id = db.Column(db.Integer, primary_key=True, comment="ID")
username = db.Column(db.String(50), comment="username")
password = db.Column(db.String(50), comment="password")
email = db.Column(db.String(50), comment="邮箱")
def __str__(self):
return self.username
# 1. 方法一： 类的继承
class AddUser(Command):
# 添加注释，标明使用功能
"""添加用户"""
def run(self):
u = User(username="fentiao90009", password="fentiao", email="fentiao@qq.com")
db.session.add(u)
db.session.commit()
print("正在添加用户%s........" % u.username)
# select
userList = User.query.all()
for item in userList:
print(item)
return True
if __name__ == "__main__":
"""
使用方法:
只是输出: python demo.py print
MySQL数据库中添加数据: python demo.py adduser
"""
manager = Manager(app)
manager.add_command("print", Print())
manager.add_command("adduser", AddUser())
manager.run()
示例三：oracle连接需要指定instant_client
# 1. 初始化 Oracle instantclient
import cx_Oracle
# instantclient_lib = "/Users/mac/Documents/www_python/wannet/instantclient_basic/mac"
if "darwin" in sys.platform.lower():
instantclient_lib = os.path.join(os.path.abspath("../"), "instantclient_basic/mac")
elif "win" in sys.platform.lower():
instantclient_lib = os.path.join(os.path.abspath("../"), "instantclient_basic/win")
else:
instantclient_lib = os.path.join(os.path.abspath("../"), "instantclient_basic/linux")
cx_Oracle.init_oracle_client(lib_dir=instantclient_lib)
oracle需要下载 instant_client
instant_client 目录结构
zip原始文件来源地址: https://www.oracle.com/cn/database/technologies/instant-client.html
示例四: mysql oracle 共存(多库连接)
import os
import sys
# sys.path.append(os.path.realpath("../"))
# sys.path.append("/Users/mac/Documents/www_python/wannet/applications")
from flask import Flask
from flask_sqlalchemy import SQLAlchemy
from flask_script import Command, Manager
def get_conn_str(bind):
"""
初始化连接
@:param bind 取值 mysql oracle
:return:
"""
conn_str = ""
if bind.lower() == "mysql":
conn_str = "mysql+pymysql://root:root@localhost/wannet"
elif bind.lower() == "oracle":
# 1. 初始化 Oracle instantclient
import cx_Oracle
# instantclient_lib = "/Users/mac/Documents/www_python/wannet/instantclient_basic/mac"
if "darwin" in sys.platform.lower():
instantclient_lib = os.path.join(os.path.abspath("../"), "instantclient_basic/mac")
elif "win" in sys.platform.lower():
instantclient_lib = os.path.join(os.path.abspath("../"), "instantclient_basic/win")
else:
instantclient_lib = os.path.join(os.path.abspath("../"), "instantclient_basic/linux")
cx_Oracle.init_oracle_client(lib_dir=instantclient_lib)
os.environ["NLS_LANG"] = "GERMAN_GERMANY.UTF8" # 解决中文乱码
# conn_string='oracle+cx_oracle://用户名:密码@hostIP:端口号/SID'
# 2. 配置 oracle 数据库连接
conn_str = "oracle+cx_oracle://dev:dev@localhost:1521/helowin"
else:
print("参数错误: bind")
return conn_str
app = Flask(__name__)
# 配置数据库连接方法一: 单库
# MySQL 连接
# app.config['SQLALCHEMY_DATABASE_URI'] = 'mysql+pymysql://root:root@localhost/wannet'
# Oracle 连接
# app.config['SQLALCHEMY_DATABASE_URI'] = "oracle+cx_oracle://dev:dev@localhost:1521/helowin"
# app.config['SQLALCHEMY_DATABASE_URI'] = get_conn_str(bind="mysql")
# 配置数据库连接方法二: 多库
SQLALCHEMY_BINDS = {
'oracle': get_conn_str(bind="oracle"),
'mysql': get_conn_str(bind="mysql"),
}
app.config['SQLALCHEMY_BINDS'] = SQLALCHEMY_BINDS
# 额外配置
app.config['SQLALCHEMY_COMMIT_ON_TEARDOWN'] = True
app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = False
db = SQLAlchemy(app)
class Print(Command):
"""
只是测试输出的内容
"""
def run(self):
print("hello")
class EIP(db.Model):
"""
Oracle Model
"""
__bind_key__ = 'oracle' # 已设置__bind_key__,则采用设置的数据库引擎
__tablename__ = "EIP"
ID = db.Column(db.Integer, primary_key=True, comment="ID")
NAME = db.Column(db.String(50), comment="username")
TITLE = db.Column(db.String(50), comment="password")
def __str__(self):
return self.NAME
class User(db.Model):
"""
MySQL Model
"""
__bind_key__ = 'mysql' # 已设置__bind_key__,则采用设置的数据库引擎
__tablename__ = 'user'
id = db.Column(db.Integer, primary_key=True, comment="ID")
username = db.Column(db.String(50), comment="username")
password = db.Column(db.String(50), comment="password")
email = db.Column(db.String(50), comment="邮箱")
def __str__(self):
return self.username
# 1. 方法一： 类的继承
class AddUser(Command):
# 添加注释，标明使用功能
"""添加用户"""
def run(self):
u = User(username="fentiao90009", password="fentiao", email="fentiao@qq.com")
db.session.add(u)
db.session.commit()
print("正在添加用户%s........" % u.username)
# select
userList = User.query.all()
for item in userList:
print(item)
return True
class AddDEV(Command):
# 添加注释，标明使用功能
"""添加DEV"""
def run(self):
u = EIP(ID="90009", NAME="fentiao", TITLE="fentiao@qq.com")
db.session.add(u)
db.session.commit()
print("正在添加 DEV %s........" % u.NAME)
return True
if __name__ == "__main__":
"""
使用方法:
只是输出: python demo.py print
MySQL数据库中添加数据: python demo.py adduser
Oracle数据库中添加数据: python demo.py adddev
"""
manager = Manager(app)
manager.add_command("print", Print())
manager.add_command("adduser", AddUser())
manager.add_command("adddev", AddDEV())
manager.run()
注意:
__bind_key__ = 'mysql' # 已设置__bind_key__,则采用设置的数据库引擎 用于区分这个model是哪个DB的
扩展
docker部署oracle参考: https://www.cnblogs.com/tiankx/p/14015851.html