python sqlalthemy 總結

 orm 數據狀態的預知識

瞬時狀態：剛建立的對象尚未被Session持久化、緩存中不存在這個對象的數據而且數據庫中沒有這個對象對應的數據爲瞬時狀態這個時候是沒有OID。　　　

持久狀態：對象通過Session持久化操做，緩存中存在這個對象的數據爲持久狀態而且數據庫中存在這個對象對應的數據爲持久狀態這個時候有OID。
遊離狀態：當Session關閉，緩存中不存在這個對象數據而數據庫中有這個對象的數據而且有OID爲遊離狀態。

數據實例對象有四種狀態，分別是
Transient - （瞬時的）
表示該實例對象不在session中，固然也沒有保存到數據庫中，
主鍵通常狀況下爲None（若是一個Persistent狀態的對象進行事務回滾後雖然主鍵有值，但倒是Transient狀態）。

Pending - （掛起的）
調用session.add()後，Transient狀態的對象就會變成Pending狀態的對象，這個時候它只是在session中，
並無保存到數據庫，所以主鍵依舊爲None。
只有觸發了session.flush()操做纔會保存到數據庫使得主鍵有值，好比查詢操做就會觸發flush。

Persistent - （持久的）
session和數據庫中都有對應的記錄存在，爲持久狀態。

Detached - （遊離的）
數據庫中可能有記錄，可是session中不存在。對這種對象的操做不會觸發任何SQL語句。

 

文檔 http://docs.sqlalchemy.org/en/latest/
#------------- connect db two kinds or many , and finally it tests for pass for every platform!
#------------- mapping bean to db.table
#------------- orm crud
#------------- sql operation and compex sql

#------------- connect db two kinds or many , and finally it can pass for every platform!
def connectDBgetSesion():
engine = create_engine('sqlite:///db.sqlite3', echo=True) # echo=False 爲不顯示sql
# engine = create_engine("mysql+pymysql://root:12456@localhost:3306/tessql?charset=utf8", echo=True)

Session = sessionmaker(bind=engine)
session = Session()
return session
def close(session):
session.commit()
session.close()
#------------- mapping bean to db.table
engine = create_engine('sqlite:///../db.sqlite3', echo=True)
# engine = create_engine("mysql+pymysql://root:123456@localhost:3306/testpymysql?charset=utf8", echo=True)
Base = declarative_base()

class User(Base):
__tablename__ = 'users'
# id = Column(Integer, primary_key=True)
id = Column(Integer, Sequence('user_id_seq'), primary_key=True)
name = Column(String(50))
fullname = Column(String(50))
password = Column(String(12))

def __repr__(self):
return "<User(name='%s', fullname='%s', password='%s')>" % (self.name, self.fullname, self.password)

# Base.metadata.create_all(engine)

#------------- orm crud
#-------------c
try:
session = connectDBgetSesion()
# "INSERT INTO custom__desc ( ,word_desc) VALUES (%s,%s) "
custom_ _desc = Custom_ _desc( =new word,word_desc=newperfword)
session.add(custom_ _desc)
except Exception as e:
print(e)
finally:
close(session)
#-------------r
# sql = "SELECT DISTINCT word_desc FROM t"
all_custom_ = session.query(distinct(t.word_desc)).all()

條件
# sql = "SELECT id,ds_desc FROM diais where diaed is NULL GROUP BY id DESC LIMIT 0,1"
rows = session.query(Di.id, Dit.dio).filter(Diaat.diagnosed == None).order_by(desc(Dt.id)).first()

# sql = "SELECT i,c FROM ic ORDER BY LENGTH(wosc) DESC" 能夠用char_length來計算字符長度，代替sql 的length長度
standard dict = session.query(I.d, Icc.woc).order_by(
desc(char_length(Ic.wc))).all()

# SELECT * from ic where LIKE %s LIMIT 0,10" like 的用法
rows1 = session.query( d_desc).filter(I.d.like("%" + nrd + "%")).offset(0).limit(12).all()
dict1 = {}
for row in rows1:
row = row._asdict() # tuple 轉字典
= row[' ']
old value = dict1.get( )

User.query.filter(User.email.endswith('@example.com')).all()

reverse = session.query(cd_desc).filter(cd_desc. .in_(rlist)).all()

不在列表中(not in)， 例如query.filter(~name.in_(['Micheal', 'Bob', 'Jack']))
空值(null), 例如 query.filter(name == None)

與(and), 例如 query.filter(and_(name == 'Andy', fullname == 'Andy Liu' ))

and_能夠省略， 例如 query.filter(name=='Andy', fullname==‘Andy Liu')
或(or), 例如 query.filter(or_(name == 'Andy', name == 'Micheal'))

#-------------u
多個更新
session.query(Diat).filter(Diat.id == dsid).update({.mad: reverse s, .dia: 1})
session.rollback()
#-------------d

#------------- sql operation and compex sql

# 聚合查詢print session.query(func.count('*')).select_from(User).scalar()print session.query(func.count('1')).select_from(User).scalar()print session.query(func.count(User.id)).scalar()print session.query(func.count('*')).filter(User.id > 0).scalar() # filter() 中包含 User，所以不須要指定表print session.query(func.count('*')).filter(User.name == 'a').limit(1).scalar() == 1 # 能夠用 limit() 限制 count() 的返回數print session.query(func.sum(User.id)).scalar()print session.query(func.now()).scalar() # func 後能夠跟任意函數名，只要該數據庫支持print session.query(func.current_timestamp()).scalar()print session.query(func.md5(User.name)).filter(User.id == 1).scalar()