ORM 框架
ORM 框架
轉:https://www.cnblogs.com/pycode/p/mysql-orm.html
SQLAlchemy是Python編程語言下的一款ORM框架,該框架建立在數據庫API之上,使用關系對象映射進行數據庫操作,簡言之便是:將對象轉換成SQL,然后使用數據API執行SQL并獲取執行結果。
SQLAlchemy本身無法操作數據庫,其必須依賴pymsql等第三方插件,Dialect用于和數據API進行交流,根據配置文件的不同調用不同的數據庫API,從而實現對數據庫的操作,如:
MySQL-Python
mysql+mysqldb://<user>:<password>@<host>[:<port>]/<dbname>
pymysql
mysql+pymysql://<username>:<password>@<host>/<dbname>[?<options>]
MySQL-Connector
mysql+mysqlconnector://<user>:<password>@<host>[:<port>]/<dbname>
cx_Oracle
oracle+cx_oracle://user:pass@host:port/dbname[?key=value&key=value...]
底層處理
使用 Engine/ConnectionPooling/Dialect 進行數據庫操作,Engine使用ConnectionPooling連接數據庫,然后再通過Dialect執行SQL語句。
from sqlalchemy import create_engine
#創建引擎
engine = create_engine("mysql+pymysql://lx:[email protected]:3306/Role", max_overflow=5)
#執行sql語句
engine.execute("INSERT INTO user (name) VALUES ('lx')")
result = engine.execute('select * from user')
res = result.fetchall()
print(res)
ORM功能使用
使用 ORM/Schema Type/SQL Expression Language/Engine/ConnectionPooling/Dialect 所有組件對數據進行操作。根據類創建對象,對象轉換成SQL,執行SQL。
創建表
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import Column, Integer, String, ForeignKey, UniqueConstraint, Index
from sqlalchemy.orm import sessionmaker, relationship
from sqlalchemy import create_engine
engine = create_engine("mysql+pymysql://root:[email protected]:3306/lx?charset=utf8", max_overflow=5,encoding='utf-8')
Base = declarative_base()
# 創建單表
class Users(Base):
__tablename__ = 'users'
id = Column(Integer, primary_key=True)
name = Column(String(32))
extra = Column(String(16))
# 一對多
class Favor(Base):
__tablename__ = 'favor'
nid = Column(Integer, primary_key=True)
caption = Column(String(50), default='red', unique=True)
class Person(Base):
__tablename__ = 'person'
nid = Column(Integer, primary_key=True)
name = Column(String(32), index=True, nullable=True)
favor_id = Column(Integer, ForeignKey("favor.nid"))
# 多對多
class ServerToGroup(Base):
__tablename__ = 'servertogroup'
nid = Column(Integer, primary_key=True, autoincrement=True)
server_id = Column(Integer, ForeignKey('server.id'))
group_id = Column(Integer, ForeignKey('group.id'))
class Group(Base):
__tablename__ = 'group'
id = Column(Integer, primary_key=True)
name = Column(String(64), unique=True, nullable=False)
class Server(Base):
__tablename__ = 'server'
id = Column(Integer, primary_key=True, autoincrement=True)
hostname = Column(String(64), unique=True, nullable=False)
port = Column(Integer, default=22)
Base.metadata.create_all(engine) #創建表
# Base.metadata.drop_all(engine) #刪除表
Session = sessionmaker(bind=engine)
session = Session()
- 增
obj = Users(name="alex0", extra='a1')
session.add(obj)
session.add_all([
Users(name="alex1", extra='a2'),
Users(name="alex2", extra='a3')
])
session.commit()
- 刪
session.query(Users).filter(Users.id > 2).delete()
session.commit()
- 改
session.query(Users).filter(Users.id > 2).update({"name" : "099"})
session.query(Users).filter(Users.id > 2).update({Users.name: Users.name + "099"}, synchronize_session=False)
session.query(Users).filter(Users.id > 2).update({"num": Users.num + 1}, synchronize_session="evaluate")
session.commit()
- 查
ret = session.query(Users).all()
ret = session.query(Users.name, Users.extra).all()
ret = session.query(Users).filter_by(name='alex').all()
ret = session.query(Users).filter_by(name='alex').first()
- 其他
# 條件
ret = session.query(Users).filter_by(name='alex').all()
ret = session.query(Users).filter(Users.id > 1, Users.name == 'eric').all()
ret = session.query(Users).filter(Users.id.between(1, 3), Users.name == 'eric').all()
ret = session.query(Users).filter(Users.id.in_([1,3,4])).all()
ret = session.query(Users).filter(~Users.id.in_([1,3,4])).all()
# 嵌套查詢
ret = session.query(Users).filter(Users.id.in_(session.query(Users.id).filter_by(name='eric'))).all()
# and_, or_
from sqlalchemy import and_, or_
ret = session.query(Users).filter(and_(Users.id > 3, Users.name == 'eric')).all()
ret = session.query(Users).filter(or_(Users.id < 2, Users.name == 'eric')).all()
ret = session.query(Users).filter(
or_(
Users.id < 2,
and_(Users.name=='eric', Users.id > 3),
Users.extra != ""
)).all()
# 通配符
ret = session.query(Users).filter(Users.name.like('e%')).all()
ret = session.query(Users).filter(~Users.name.like('e%')).all()
# 限制
ret = session.query(Users)[1:2]
# 排序
ret = session.query(Users).order_by(Users.name.desc()).all()
ret = session.query(Users).order_by(Users.name.desc(), Users.id.asc()).all()
# 分組
from sqlalchemy.sql import func
ret = session.query(Users.extra, func.count('*')).group_by(Users.extra).all()
ret = session.query(
func.max(Users.id),
func.sum(Users.id),
func.min(Users.id)).group_by(Users.name).all()
ret = session.query(
func.max(Users.id),
func.sum(Users.id),
func.min(Users.id)).group_by(Users.name).having(func.min(Users.id) >2).all()
# 連表
ret = session.query(Users, Favor).filter(Users.id == Favor.nid).all()
# 內連接
ret = session.query(Person).join(Favor).all()
# 外連接
ret = session.query(Person).join(Favor, isouter=True).all()
# 組合 union聯合查詢
q1 = session.query(Users.name).filter(Users.id > 2)
q2 = session.query(Favor.caption).filter(Favor.nid < 2)
ret = q1.union(q2).all()
q1 = session.query(Users.name).filter(Users.id > 2)
q2 = session.query(Favor.caption).filter(Favor.nid < 2)
ret = q1.union_all(q2).all()
- ORM解決中文編碼問題 sqlalchemy 默認使用latin-1進行編碼。所以當出現中文時就會報如下錯誤:
UnicodeEncodeError: 'latin-1' codec can't encode characters in position 39-41: ordinal not in range(256)
解決方法:
在連接數據庫的時候直接指定字符編碼:
engine = create_engine("mysql+pymysql://lx:[email protected]:3306/Role?charset=utf8", max_overflow=5,encoding='utf-8')
- ORM 指定查詢返回數據格式 默認使用query查詢返回的結果為一個對象
res = session.query(User).all()
print(res)
輸出結果:
[<__main__.User object at 0x10385c438>, <__main__.User object at 0x10385c4a8>, <__main__.User object at 0x10385c550>, <__main__.User object at 0x10385c5f8>, <__main__.User object at 0x10385c6a0>]
解決方法:
使用__repr__定義返回的數據
class User(Base):
__tablename__ = 'user'
nid = Column(Integer,primary_key=True,autoincrement=True)
name = Column(String(10),nullable=False)
def __repr__(self):
output = "(%s,%s,%s)" %(self.nid,self.name,self.role)
return output
res = session.query(User).all()
print(res)
輸出:
[(1,fuzj,1), (2,jie,2), (3,張三,2), (4,李四,1), (5,王五,3)]
一對多
實例
- ORM一對多具體使用
- 設計兩張表,user表和role表
- user 表中存放用戶,role表中存放用戶角色,role表中角色對應user表中多個用戶,user表中一個用戶只對應role表中一個角色,中間通過外鍵約束
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import Column, Integer, String,ForeignKey
from sqlalchemy.orm import sessionmaker,relationship
from sqlalchemy import create_engine
engine = create_engine("mysql+pymysql://root:[email protected]:3306/lx02?charset=utf8", max_overflow=5,encoding='utf-8')
Base = declarative_base()
class Role(Base):
__tablename__ = 'role'
rid = Column(Integer, primary_key=True, autoincrement=True) #主鍵,自增
role_name = Column(String(10))
def __repr__(self):
output = "(%s,%s)" %(self.rid,self.role_name)
return output
class User(Base):
__tablename__ = 'user'
nid = Column(Integer,primary_key=True,autoincrement=True)
name = Column(String(10),nullable=False)
role_id = Column(Integer,ForeignKey('role.rid')) #外鍵關聯
def __repr__(self):
output = "(%s,%s,%s)" %(self.nid,self.name,self.role_id)
return output
Base.metadata.create_all(engine)
Session = sessionmaker(bind=engine)
session = Session()
#添加角色數據
session.add(Role(role_name='dba'))
session.add(Role(role_name='sa'))
session.add(Role(role_name='net'))
#添加用戶數據
session.add_all([
User(name='小紅',role='1'),
User(name='小明',role='2'),
User(name='張三',role='2'),
User(name='李四',role='1'),
User(name='王五',role='3'),
])
session.commit()
session.close()
- 普通連表查詢
res = session.query(User,Role).join(Role).all() #查詢所有用戶,及對應的role id
res1 = session.query(User.name,Role.role_name).join(Role).all() #查詢所有用戶和角色,
res2 = session.query(User.name,Role.role_name).join(Role,isouter=True).filter(Role.role_name=='sa').all() #查詢所有sa的用戶
print(res)
print(res1)
print(res2)
輸出結果:
[((1,小紅,1), (1,dba)), ((2,小明,2), (2,sa)), ((3,張三,2), (2,sa)), ((4,李四,1), (1,dba)), ((5,王五,3), (3,net))]
[('小紅', 'dba'), ('小明', 'sa'), ('張三', 'sa'), ('李四', 'dba'), ('王五', 'net')]
[('jie', 'sa'), ('張三', 'sa')]
使用relationship 添加影射關系進行查詢(反向查詢)
- 首先在User表中添加relationship映射關系
class Role(Base):
__tablename__ = 'role'
rid = Column(Integer, primary_key=True, autoincrement=True) #主鍵,自增
role_name = Column(String(10))
def __repr__(self):
output = "(%s,%s)" %(self.rid,self.role_name)
return output
class User(Base):
__tablename__ = 'user'
nid = Column(Integer,primary_key=True,autoincrement=True)
name = Column(String(10),nullable=False)
role_id = Column(Integer,ForeignKey('role.rid'))
role = relationship("Role",backref='user') #Role為類名
def __repr__(self):
output = "(%s,%s,%s)" %(self.nid,self.name,self.role)
return output
- 查詢
#正向查詢
print('正向查詢')
res = session.query(User).all() #查詢所有的用戶和角色
for u in res:
print(u.name,u.role.role_name) #此時的u.role 就是role表對應的關系
res = session.query(User).filter(User.name=='小明').first() #查詢fuzj用戶和角色
print(res.name,res.role.role_name)
print('反向查找')
#反向查找
res = session.query(Role).filter(Role.role_name =='dba').first() #查找dba組下的所有用戶
print(res.user) #此時 print的結果為[(1,小明,1), (4,李四,1)]
for i in res.user:
print(i.name,res.role_name)
輸出結果:
正向查詢
小明 dba
小紅 sa
張三 sa
李四 dba
王五 net
小明 dba
反向查找
[(1,小明,1), (4,李四,1)]
小明 dba
李四 dba
- 說明
relationship 在user表中創建了新的字段,這個字段只用來存放user表中和role表中的對應關系,在數據庫中并不實際存在
所謂正向和反向查找是對于relationship關系映射所在的表而說,如果通過該表(user表)去查找對應的關系表(role表),就是正向查找,反正通過對應的關系表(role表)去查找該表(user表)即為反向查找。而relationship往往會和ForeignKey共存在一個表中。
ORM多對多
實例
Mysql多對多關系指的是兩張表A和B本沒有任何關系,而是通過第三張表C建立關系,通過關系表C,使得表A在表B中存在多個關聯數據,表B在表A中同樣存在多個關聯數據
- 創建三張表 host表 hostuser表 host_to_hostuser表
- host表中存放主機,hostuser表中存放主機的用戶, host_to_hostuser表中存放主機用戶對應的主機,hostuser表中用戶對應host表中多個主機,host表中主機對應hostuser表中多個用戶,中間關系通過host_to_hostuser表進行關聯。host_to_hostuser和host表、user表進行外鍵約束
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import Column, Integer, String,ForeignKey
from sqlalchemy.orm import sessionmaker,relationship
from sqlalchemy import create_engine
engine = create_engine("mysql+pymysql://root:[email protected]:3306/lx03?charset=utf8", max_overflow=5,encoding='utf-8')
Base = declarative_base()
class Host(Base):
__tablename__ = 'host'
nid = Column(Integer, primary_key=True,autoincrement=True)
hostname = Column(String(32))
port = Column(String(32))
ip = Column(String(32))
class HostUser(Base):
__tablename__ = 'host_user'
nid = Column(Integer, primary_key=True,autoincrement=True)
username = Column(String(32))
class HostToHostUser(Base):
__tablename__ = 'host_to_host_user'
nid = Column(Integer, primary_key=True,autoincrement=True)
host_id = Column(Integer,ForeignKey('host.nid'))
host_user_id = Column(Integer,ForeignKey('host_user.nid'))
Base.metadata.create_all(engine)
Session = sessionmaker(bind=engine)
session = Session()
#添加數據
session.add_all([
Host(hostname='c1',port='22',ip='1.1.1.1'),
Host(hostname='c2',port='22',ip='1.1.1.2'),
Host(hostname='c3',port='22',ip='1.1.1.3'),
Host(hostname='c4',port='22',ip='1.1.1.4'),
Host(hostname='c5',port='22',ip='1.1.1.5'),
])
session.add_all([
HostUser(username='root'),
HostUser(username='db'),
HostUser(username='nb'),
HostUser(username='sb'),
])
session.add_all([
HostToHostUser(host_id=1,host_user_id=1),
HostToHostUser(host_id=1,host_user_id=2),
HostToHostUser(host_id=1,host_user_id=3),
HostToHostUser(host_id=2,host_user_id=2),
HostToHostUser(host_id=2,host_user_id=4),
HostToHostUser(host_id=2,host_user_id=3),
])
session.commit()
session.close()
- 普通多次查詢
第一步:host_id = session.query(Host.nid).filter(Host.hostname=='c2').first() #查找hostbane對應的hostid,返回結果為元組(2,)
第二步: user_id_list = session.query(HostToHostUser.host_user_id).filter(HostToHostUser.host_id==host_id[0]).all() #查詢hostid對應的所有userid
user_id_list = zip(*user_id_list) #user_id_list 初始值為[(2,), (4,), (3,)],使用zip轉換為[2,4,3]對象
#print(list(user_id_list)) #結果為[(2, 4, 3)]
第三步:user_list = session.query(HostUser.username).filter(HostUser.nid.in_(list(user_id_list)[0])).all() #查詢符合條件的用戶
print(user_list)
#或者:
user_id_list = session.query(HostToHostUser.host_user_id).join(Host).filter(Host.hostname=='c2').all()
user_id_list = zip(*user_id_list)
user_list = session.query(HostUser.username).filter(HostUser.nid.in_(list(user_id_list)[0])).all()
print(user_list)
# 輸出結果
[('db',), ('nb',), ('sb',)]
使用relationship映射關系查詢
- 首先在關系表Host_to_hostuser中加入relationship關系映射
class HostToHostUser(Base):
__tablename__ = 'host_to_host_user'
nid = Column(Integer, primary_key=True,autoincrement=True)
host_id = Column(Integer,ForeignKey('host.nid'))
host_user_id = Column(Integer,ForeignKey('host_user.nid'))
host = relationship('Host',backref='h') #對應host表
host_user = relationship('HostUser',backref='u') #對應host_user表
- 查詢
#查找一個服務器上有哪些用戶
res = session.query(Host).filter(Host.hostname=='c2').first() #返回的是符合條件的服務器對象
res2 = res.h #通過relationship反向查找 Host_to_Hostuser中的對應關系
for i in res2: #i為host_to_hostuser表和host表中c2主機有對應關系的條目
print(i.host_user.username) #正向查找, 通過relationship ,找到host_to_hostuser中對應的hostuser 即i.host_user
#查找此用戶有哪些服務器
res = session.query(HostUser).filter(HostUser.username=='sb').first()
for i in res.u:
print(i.host.hostname)
擴展查詢
- 不查詢關系表(中間表),直接在hostuser表中指定關系表,然后獲取host表
- 在host表中使用 relationship的secondary指定關系表。
class Host(Base):
__tablename__ = 'host'
nid = Column(Integer, primary_key=True,autoincrement=True)
hostname = Column(String(32))
port = Column(String(32))
ip = Column(String(32))
host_user = relationship('HostUser',secondary=lambda :HostToHostUser.__table__,backref='h')
查詢:
host_obj = session.query(Host).filter(Host.hostname=='c1').first()
for i in host_obj.host_user:
print(i.username)
創建一個實時監控文件系統
我將使用 SQLAlchemy 演示如何創建一個元數據工具。此工具的目標是監控文件系統、創建和刪除事件,以及在一個 SQLAlchemy 數據庫中保存這些變更的記錄
#/usr/bin/env python2.7
import os
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import Table, Column, Integer, String, MetaData, ForeignKey
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from sqlalchemy.orm import scoped_session
from pyinotify import *
path = "/tmp"
#SQLAlchemy
engine = create_engine('sqlite:///meta.db', echo=True)
Base = declarative_base()
Session = scoped_session(sessionmaker(bind=engine))
class Filesystem(Base):
__tablename__ = 'filesystem'
path = Column(String, primary_key=True)
name = Column(String)
def __init__(self, path,name):
self.path = path
self.name = name
def __repr__(self):
return "<Metadata('%s','%s')>" % (self.path,self.name)
# 裝飾器
def transactional(fn):
"""add transactional semantics to a method."""
def transact(self, *args):
session = Session()
try:
fn(self, session, *args)
session.commit()
except:
session.rollback()
raise
transact.__name__ = fn.__name__
return transact
class ProcessDir(ProcessEvent):
"""Performs Actions based on mask values"""
@transactional
def process_IN_CREATE(self, session, event):
print "Creating File and File Record:", event.pathname
create_record = Filesystem(event.pathname, event.path)
session.add(create_record)
@transactional
def process_IN_DELETE(self, session, event):
print "Removing:", event.pathname
delete_record = session.query(Filesystem).\
filter_by(path=event.pathname).one()
session.delete(delete_record)
def init_repository():
#Drop the table, then create again with each run
Base.metadata.drop_all(engine)
Base.metadata.create_all(engine)
session = Session()
#Initial Directory Walking Addition Brute Force
for dirpath, dirnames, filenames in os.walk(path):
for file in filenames:
fullpath = os.path.join(dirpath, file)
record = Filesystem(fullpath, file)
session.add(record)
session.flush()
for record in session.query(Filesystem):
print "Database Record Number: Path: %s , File: %s " \
% (record.path, record.name)
session.commit()
if __name__ == "__main__":
init_repository()
#Pyionotify
wm = WatchManager()
mask = IN_DELETE | IN_CREATE
notifier = ThreadedNotifier(wm, ProcessDir())
notifier.start()
wdd = wm.add_watch(path, mask, rec=True)
智能推薦
ORM持久化框架MyBatis
一MyBatis的介紹 MyBatis 是一個支持普通SQL查詢,存儲過程和高級映射的優秀持久層框架。 MyBatis 消除了幾乎所有的JDBC代碼和手工設置參數以及結果集的檢索。 MyBatis 使用簡單的 XML或注解用于配置和原始映射,將接口和 Java 的POJOs(Plain Old Java Objects,普通的 Java對象)映射成數據庫中的記錄。 MyBatis不能自動創建表,如...
ORM框架之Mybatis(二)
一、MybatisMybatis連接池與事務深入 1.1 Mybatis的連接池技術 在Mybatis中也有連接池技術,但是它采用的是自己的連接池技術。在Mybatis的SqlMapConfig.xml配置文件中,通過<dataSource type=”pooled”>來實現Mybatis中連接池的配置 1.1.1 Mybatis連接池的...
ORM框架sqlalchemy的使用
文章目錄 前言 SQLALchemy 1.引擎設置(Engine Configuration) 2.聲明映射類型 基類的創建 實體類的創建 3.創建映射類的實例 創建和刪除表 4.創建會話session 5.CURD操作 實體的狀態 增 修改 刪除 查詢 1.條件查詢 2.排序 3.分頁 4.消費者方法 5.聚合分組 6.關聯查詢 前言 SQLALchemy對象關系映射器提供了一種方法,用于將用戶...
輕量級ORM框架---JOOQ
當今企業級開發中,幾乎都會用到orm框架,有全自動的orm框架hibernate,也有半自動的orm框架mybatis,還有springboot推薦使用的jpa,作者本人個偏向于mybatis框架(寫原生sql真的很爽啊!!!),但是在高并發的場景下,不推薦在sql語句中使用關聯查詢,更傾向于單張表表冗余查詢,在代碼中使用stream操作,當然這是題外話。前幾天逛論壇意外發現一篇標題為《拋棄myb...
Android ORM 框架之 greenDAO
概要 GreenDao簡介 我相信,在平時的開發過程中,大家一定會或多或少地接觸到 SQLite。然而在使用它時,我們往往需要做許多額外的工作,像編寫 SQL 語句與解析查詢結果等。所以,適用于 Android 的ORM 框架也就孕育而生,現在市面上主流的框架有 OrmLite、SugarORM、Active Android、Realm 與 GreenDAO。 greenDAO是一種Android...
猜你喜歡
freemarker + ItextRender 根據模板生成PDF文件
1. 制作模板 2. 獲取模板,并將所獲取的數據加載生成html文件 2. 生成PDF文件 其中由兩個地方需要注意,都是關于獲取文件路徑的問題,由于項目部署的時候是打包成jar包形式,所以在開發過程中時直接安照傳統的獲取方法沒有一點文件,但是當打包后部署,總是出錯。于是參考網上文章,先將文件讀出來到項目的臨時目錄下,然后再按正常方式加載該臨時文件; 還有一個問題至今沒有解決,就是關于生成PDF文件...
電腦空間不夠了?教你一個小秒招快速清理 Docker 占用的磁盤空間!
Docker 很占用空間,每當我們運行容器、拉取鏡像、部署應用、構建自己的鏡像時,我們的磁盤空間會被大量占用。 如果你也被這個問題所困擾,咱們就一起看一下 Docker 是如何使用磁盤空間的,以及如何回收。 docker 占用的空間可以通過下面的命令查看: TYPE 列出了docker 使用磁盤的 4 種類型: Images:所有鏡像占用的空間,包括拉取下來的鏡像,和本地構建的。 Con...
requests實現全自動PPT模板
http://www.1ppt.com/moban/ 可以免費的下載PPT模板,當然如果要人工一個個下,還是挺麻煩的,我們可以利用requests輕松下載 訪問這個主頁,我們可以看到下面的樣式 點每一個PPT模板的圖片,我們可以進入到詳細的信息頁面,翻到下面,我們可以看到對應的下載地址 點擊這個下載的按鈕,我們便可以下載對應的PPT壓縮包 那我們就開始做吧 首先,查看網頁的源代碼,我們可以看到每一...