# sqlite 創(chuàng)建數(shù)據(jù)庫(kù)連接
engine = create_engine('sqlite:///database.db', echo=False)

# sqlite 創(chuàng)建內(nèi)存數(shù)據(jù)庫(kù)
engine = create_engine('sqlite://')
engine = create_engine('sqlite:///:memory:', echo=True)

# PostgreSQL 創(chuàng)建數(shù)據(jù)庫(kù)連接
engine = create_engine('postgresql://scott:tiger@localhost/mydatabase')                # default
engine = create_engine('postgresql+psycopg2://scott:tiger@localhost/mydatabase')       # psycopg2
engine = create_engine('postgresql+pg8000://scott:tiger@localhost/mydatabase')         # pg8000

# MySQL 創(chuàng)建數(shù)據(jù)庫(kù)連接
engine = create_engine('mysql://scott:tiger@localhost/foo')                  # default
engine = create_engine('mysql+mysqldb://scott:tiger@localhost/foo')          # mysql-python
engine = create_engine('mysql+mysqlconnector://scott:tiger@localhost/foo')   # MySQL-connector-python
engine = create_engine('mysql+oursql://scott:tiger@localhost/foo')           # OurSQL

# Oracle 創(chuàng)建數(shù)據(jù)庫(kù)連接
engine = create_engine('oracle://scott:tiger@127.0.0.1:1521/sidname')
engine = create_engine('oracle+cx_oracle://scott:tiger@tnsname')

# MSSQL 創(chuàng)建數(shù)據(jù)庫(kù)連接
engine = create_engine('mssql+pyodbc://scott:tiger@mydsn')                   # pyodbc
engine = create_engine('mssql+pymssql://scott:tiger@hostname:port/dbname')   # pymssql

import sqlite3,datetime,time
from sqlalchemy.orm import sessionmaker
from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import Column, Integer, String, Float, DateTime, Date, Time, Boolean, DECIMAL, Enum, Text

# 建立基本映射類(lèi)
Base = declarative_base()

# 創(chuàng)建SQLITE數(shù)據(jù)庫(kù)
engine = create_engine('sqlite:///:memory:', echo=False)

# 創(chuàng)建映射類(lèi)User
class User(Base):
    __tablename__ = 'UserDB'

    # 主鍵 primary_key | 自動(dòng)增長(zhǎng) autoincrement | 不為空 nullable | 唯一性約束 unique
    id = Column(Integer, primary_key=True, autoincrement=True, nullable=True, unique=True)

    # 字符串類(lèi)型
    username = Column(String(32), nullable=True, default="none")
    password = Column(String(32), nullable=True, default="none")

    # 姓名字段默認(rèn)值是0
    age = Column(Integer,nullable=False, default=0)

    # 增加創(chuàng)建日期 [日期:時(shí)間]
    create_time = Column(DateTime, default=datetime.datetime.now)

    # onupdate=datetime.now 每次更新數(shù)據(jù)的時(shí)候都要更新該字段值
    update_time = Column(DateTime, onupdate=datetime.datetime.now, default=datetime.datetime.now)

    # 增加用戶分?jǐn)?shù)
    user_value = Column(Float, default=0.0)

    # 枚舉類(lèi)型定義
    # tag = Column(Enum("python",'flask','django'))

    # __repr__方法用于輸出該類(lèi)的對(duì)象被print()時(shí)輸出的字符串，如果不想寫(xiě)可以不寫(xiě)
    def __repr__(self):
        return "<UserDB(username='%s',password='%s')>" % (self.username,self.password)

if __name__ == "__main__":
    print("當(dāng)前表名: {}".format(User.__table__))

    # 創(chuàng)建會(huì)話
    Session = sessionmaker(bind=engine)
    session = Session()

    # 創(chuàng)建數(shù)據(jù)表
    Base.metadata.create_all(engine, checkfirst=True)

    # 逐條增加新記錄
    insert_user = User(username='lyshark', password='123456', age=24, user_value=12.5)
    session.add(insert_user)

    insert_user = User(username='sqlalchemy', password='123', age=34, user_value=45.8)
    session.add(insert_user)

    # 插入多條記錄
    session.add_all(
        [
         User(username="admin", password="123123", age=54, user_value=66.9),
         User(username="root", password="3456576", age=67, user_value=98.4)
        ]
    )

    # 提交事務(wù)
    session.commit()

import sqlite3,time,datetime
from sqlalchemy import func
from sqlalchemy import or_
from sqlalchemy.orm import sessionmaker
from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import Column, Integer, String, Float,DateTime

# 建立基本映射類(lèi)
Base = declarative_base()

# 創(chuàng)建SQLITE數(shù)據(jù)庫(kù)
engine = create_engine('sqlite:///:memory:', echo=False)

# 創(chuàng)建映射類(lèi)User
class User(Base):
    __tablename__ = 'UserDB'
    id = Column(Integer, primary_key=True, autoincrement=True, nullable=True, unique=True)
    username = Column(String(32), nullable=True, default="none")
    password = Column(String(32), nullable=True, default="none")
    age = Column(Integer,nullable=False, default=0)
    create_time = Column(DateTime, default=datetime.datetime.now)
    update_time = Column(DateTime, onupdate=datetime.datetime.now, default=datetime.datetime.now)
    user_value = Column(Float, default=0.0)

    # __repr__方法用于輸出該類(lèi)的對(duì)象被print()時(shí)輸出的字符串，如果不想寫(xiě)可以不寫(xiě)
    def __repr__(self):
        return "<UserDB(username='%s',password='%s')>" % (self.username,self.password)

if __name__ == "__main__":
    # 創(chuàng)建會(huì)話
    Session = sessionmaker(bind=engine)
    session = Session()

    # 創(chuàng)建數(shù)據(jù)表(存在則跳過(guò))
    Base.metadata.create_all(engine, checkfirst=True)

    # 查詢所有字段
    all_value = session.query(User).all()
    for item in all_value:
        print("ID: {} --> 用戶: {}".format(item.id, item.username))

    # 查詢指定字段
    key_value = session.query(User.username,User.password).all()
    print(key_value)

    # 查詢第一條
    first_value = session.query(User).first()
    print("第一條記錄: {} {}".format(first_value.username, first_value.password))

    # 使用過(guò)濾器 [ 過(guò)濾出age>20的用戶,輸出其(id,username)字段 ]
    filter_value = session.query(User.id,User.username).filter(User.age > 20).all()
    print("過(guò)濾結(jié)果: {}".format(filter_value))

    # 排序輸出 [ 正序/倒序 ]
    sort_value = session.query(User.username,User.age).order_by(User.age).all()
    print("正序排列: {}".format(sort_value))

    sort_value = session.query(User.username,User.age).order_by(User.age.desc()).all()
    print("倒序排列: {}".format(sort_value))

    # 查詢計(jì)數(shù)
    count_value = session.query(User).count()
    print("記錄條數(shù): {}".format(count_value))

    # and/or 條件過(guò)濾 默認(rèn)為and 在filter()中用,分隔多個(gè)條件表示,如果是or則需增加or_連接多個(gè)條件
    and_value = session.query(User.username,User.age).filter(User.age >= 20, User.age <= 40).all()
    print("與查詢: {}".format(and_value))

    or_value = session.query(User.username,User.age).filter(or_(User.age >= 20, User.age <= 40)).all()
    print("或查詢: {}".format(or_value))

    # 等于查詢
    equal_value = session.query(User.username,User.password).filter(User.age == 67).all()
    print("等于查詢: {}".format(equal_value))

    not_equal_value = session.query(User.username,User.password).filter(User.age != 67).all()
    print("不等于查詢: {}".format(not_equal_value))

    # like模糊匹配
    like_value = session.query(User.username,User.create_time).filter(User.username.like("%ly%")).all()
    print("模糊匹配: {}".format(like_value))

    # in查詢范圍
    in_value = session.query(User.username,User.password).filter(User.age.in_([24,34])).all()
    print("查詢兩者: {}".format(in_value))

    not_in_value = session.query(User.username,User.password).filter(User.age.notin_([24,34])).all()
    print("查詢非兩者: {}".format(not_in_value))

    # op正則匹配查詢
    op_value = session.query(User.username).filter(User.username.op("regexp")("^a")).all()
    print("正則匹配: {}".format(op_value))

    # 調(diào)用數(shù)據(jù)庫(kù)內(nèi)置函數(shù)
    func_value = session.query(func.count(User.age)).one()
    print("調(diào)用函數(shù): {}".format(func_value))

    # 數(shù)據(jù)切片
    cat_value = session.query(User.username).all()[:2]
    print("輸出前兩條: {}".format(cat_value))

    cat_value = session.query(User.username).offset(5).limit(3).all()
    print("第6行開(kāi)始顯示前3個(gè): {}".format(cat_value))

    cat_value = session.query(User.username).order_by(User.id.desc())[0:10]
    print("輸出最后10個(gè): {}".format(cat_value))

    # 非空查詢
    isnot_value = session.query(User).filter(User.username.isnot(None)).all()
    print("非空顯示: {}".format(isnot_value))

    null_value = session.query(User).filter(User.username.is_(None)).all()
    print("為空顯示: {}".format(null_value))

    # 分組測(cè)試
    group_by = session.query(User.username,func.count(User.id)).group_by(User.age).all()
    print("根據(jù)年齡分組: {}".format(group_by))

    # 進(jìn)一步過(guò)濾查詢
    having_by = session.query(User.username,User.age).group_by(User.age).having(User.age > 30).all()
    print("以age分組,并查詢age大于30的記錄: {}".format(having_by))

import sqlite3,time,datetime
from sqlalchemy import func
from sqlalchemy import or_
from sqlalchemy.orm import sessionmaker
from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import Column, Integer, String, Float,DateTime

# 建立基本映射類(lèi)
Base = declarative_base()

# 創(chuàng)建SQLITE數(shù)據(jù)庫(kù)
engine = create_engine('sqlite:///:memory:', echo=False)

# 創(chuàng)建映射類(lèi)User
class User(Base):
    __tablename__ = 'UserDB'
    id = Column(Integer, primary_key=True, autoincrement=True, nullable=True, unique=True)
    username = Column(String(32), nullable=True, default="none")
    password = Column(String(32), nullable=True, default="none")
    age = Column(Integer,nullable=False, default=0)
    create_time = Column(DateTime, default=datetime.datetime.now)
    update_time = Column(DateTime, onupdate=datetime.datetime.now, default=datetime.datetime.now)
    user_value = Column(Float, default=0.0)

    # __repr__方法用于輸出該類(lèi)的對(duì)象被print()時(shí)輸出的字符串，如果不想寫(xiě)可以不寫(xiě)
    def __repr__(self):
        return "<UserDB(username='%s',password='%s')>" % (self.username,self.password)

if __name__ == "__main__":
    # 創(chuàng)建會(huì)話
    Session = sessionmaker(bind=engine)
    session = Session()

    # 創(chuàng)建數(shù)據(jù)表(存在則跳過(guò))
    Base.metadata.create_all(engine, checkfirst=True)

    # 修改數(shù)據(jù): 先查詢?cè)谛薷?
    select_update = session.query(User).filter_by(username="lyshark").first()
    select_update.password = "test1234"
    session.commit()

    # 修改數(shù)據(jù): 直接修改
    session.query(User).filter_by(username="lyshark").update({User.password: 'abcd'})
    session.commit()

    session.query(User).filter_by(username="lyshark").update({"password": '123456'})
    session.commit()

    # 刪除數(shù)據(jù): 先查詢?cè)趧h除
    del_ptr = session.query(User).filter_by(username="lyshark").first()
    session.delete(del_ptr)
    session.commit()

    # 刪除數(shù)據(jù): 直接刪除
    session.query(User).filter(User.username=="sqlalchemy").delete()
    session.commit()

import sqlite3,time,datetime,json
from sqlalchemy.orm import sessionmaker
from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import Column, Integer, String, Float,DateTime

# 建立基本映射類(lèi)
Base = declarative_base()

# 創(chuàng)建SQLITE數(shù)據(jù)庫(kù)
engine = create_engine('sqlite:///:memory:', echo=False)

# 創(chuàng)建映射類(lèi)User
class User(Base):
    __tablename__ = 'UserDB'
    id = Column(Integer, primary_key=True, autoincrement=True, nullable=True, unique=True)
    username = Column(String(32), nullable=True, default="none")
    password = Column(String(32), nullable=True, default="none")
    age = Column(Integer,nullable=False, default=0)
    create_time = Column(DateTime, default=datetime.datetime.now)
    update_time = Column(DateTime, onupdate=datetime.datetime.now, default=datetime.datetime.now)
    user_value = Column(Float, default=0.0)

    # 查詢結(jié)果轉(zhuǎn)字典 (保留數(shù)據(jù)類(lèi)型)
    def single_to_dict(self):
        return {c.name: getattr(self, c.name) for c in self.__table__.columns}

    # 查詢結(jié)果轉(zhuǎn)字典 (全轉(zhuǎn)為字符串)
    def dobule_to_dict(self):
        result = {}
        for key in self.__mapper__.c.keys():
            if getattr(self, key) is not None:
                result[key] = str(getattr(self, key))
            else:
                result[key] = getattr(self, key)
        return result

# 將查詢結(jié)果轉(zhuǎn)為JSON
def to_json(all_vendors):
    v = [ ven.dobule_to_dict() for ven in all_vendors ]
    return v

if __name__ == "__main__":
    # 創(chuàng)建會(huì)話
    Session = sessionmaker(bind=engine)
    session = Session()

    # 創(chuàng)建數(shù)據(jù)表(存在則跳過(guò))
    Base.metadata.create_all(engine, checkfirst=True)

    # 查詢結(jié)果轉(zhuǎn)為字典(保持?jǐn)?shù)據(jù)庫(kù)格式)
    key_value = session.query(User).first()
    data = key_value.single_to_dict()
    print("轉(zhuǎn)為字典: {}".format(data))

    # 查詢結(jié)果轉(zhuǎn)為字典(字符串格式)
    key_value = session.query(User).first()
    data = key_value.dobule_to_dict()
    print("轉(zhuǎn)為字符串字典: {}".format(data))

    # 查詢結(jié)果轉(zhuǎn)為JSON格式
    key_value = session.query(User)
    data = to_json(key_value)
    print("轉(zhuǎn)為JSON格式: {}".format(data))

from werkzeug.security import generate_password_hash,check_password_hash
import sqlite3,datetime,time
from sqlalchemy.orm import sessionmaker
from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import Column, Integer, String, Float, DateTime, Date, Time, Boolean, DECIMAL, Enum, Text

# 創(chuàng)建SQLITE數(shù)據(jù)庫(kù)
engine = create_engine("sqlite:///:memory:", encoding='utf-8')
Base = declarative_base()                                       # 生成orm基類(lèi)

# 創(chuàng)建會(huì)話
Session_class = sessionmaker(bind=engine)                       # 創(chuàng)建與數(shù)據(jù)庫(kù)的會(huì)話session
session = Session_class()                                       # 生成session實(shí)例

# 創(chuàng)建映射類(lèi)User
class User(Base):
    __tablename__ = 'UserDB'
    id = Column(Integer, primary_key=True, autoincrement=True, nullable=True, unique=True)
    username = Column(String(64))
    _password_hash_ = Column(String(256))                       # 加下劃線作為私有函數(shù),無(wú)法被外部訪問(wèn)
    email = Column(String(64))

    # 設(shè)置一個(gè)password字段用來(lái)設(shè)置密碼
    @property
    def password(self):
        raise Exception("密碼不能被讀取")

    # 賦值password字段時(shí),則自動(dòng)加密存儲(chǔ)
    @password.setter
    def password(self, value):
        self._password_hash_ = generate_password_hash(value)

    # 使用 check_password,進(jìn)行密碼校驗(yàn) 返回True False。
    def check_password(self, pasword):
        return check_password_hash(self._password_hash_, pasword)

    # 設(shè)置輸出函數(shù)
    def print_function(self):
        # 密碼不可讀調(diào)用 self.password 會(huì)報(bào)錯(cuò)
        # print("用戶: {} 密碼: {}".format(self.username,self.password))

        print("用戶: {} email: {}".format(self.username, self.email))
        return True

if __name__ == "__main__":
    print("當(dāng)前表名: {}".format(User.__table__))

    # 創(chuàng)建數(shù)據(jù)表
    Base.metadata.create_all(engine, checkfirst=True)

    # 插入測(cè)試數(shù)據(jù)
    insert = User(username="lyshark",password="123123",email="lyshark@163.com")
    session.add(insert)

    insert = User(username="admin",password="556677",email="lyshark@163.com")
    session.add(insert)
    session.commit()

    # 查詢測(cè)試
    tag = session.query(User).filter_by(username="lyshark").first()
    print("測(cè)試密碼是否正確: {}".format(tag.check_password("123123")))

    # 調(diào)用函數(shù)驗(yàn)證當(dāng)前用戶
    tag = session.query(User).filter_by(username="admin").first()
    func = tag.print_function()
    print("輸出測(cè)試: {}".format(func))

import sqlite3,datetime,time
from sqlalchemy import func
from sqlalchemy.orm import sessionmaker
from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import Column, Integer, String, Float, DateTime, Date, Time, Boolean, DECIMAL, Enum, Text

# 建立基本映射類(lèi)
Base = declarative_base()

# 創(chuàng)建SQLITE數(shù)據(jù)庫(kù)
engine = create_engine('sqlite:///:memory:', echo=False)

# 創(chuàng)建映射類(lèi)User
class User(Base):
    __tablename__ = 'UserDB'

    # 主鍵 primary_key | 自動(dòng)增長(zhǎng) autoincrement | 不為空 nullable | 唯一性約束 unique
    id = Column(Integer, primary_key=True, autoincrement=True, nullable=True, unique=True)

    # 字符串類(lèi)型
    username = Column(String(32), nullable=True, default="none")
    password = Column(String(32), nullable=True, default="none")

    # 姓名字段默認(rèn)值是0
    age = Column(Integer,nullable=False, default=0)

    # 增加創(chuàng)建日期 [日期:時(shí)間]
    create_time = Column(DateTime, default=datetime.datetime.now)

    # onupdate=datetime.now 每次更新數(shù)據(jù)的時(shí)候都要更新該字段值
    update_time = Column(DateTime, onupdate=datetime.datetime.now, default=datetime.datetime.now)

    # 增加用戶分?jǐn)?shù)
    user_value = Column(Float, default=0.0)

    # 枚舉類(lèi)型定義
    # tag = Column(Enum("python",'flask','django'))

    # __repr__方法用于輸出該類(lèi)的對(duì)象被print()時(shí)輸出的字符串，如果不想寫(xiě)可以不寫(xiě)
    def __repr__(self):
        return "<UserDB(username='%s',password='%s')>" % (self.username,self.password)

if __name__ == "__main__":
    print("當(dāng)前表名: {}".format(User.__table__))

    # 創(chuàng)建會(huì)話
    Session = sessionmaker(bind=engine)
    session = Session()

    # 創(chuàng)建數(shù)據(jù)表
    Base.metadata.create_all(engine, checkfirst=True)

    # 統(tǒng)計(jì)總數(shù)
    count = session.query(func.count(User.id)).first()
    print("總記錄: {}".format(count))

    # age 字段平均值
    age_avg = session.query(func.avg(User.age)).first()
    print("平均值: {}".format(age_avg))

    # age 字段最大值
    age_max = session.query(func.max(User.age)).first()
    print("最大值: {}".format(age_max))

    # age 最小值
    age_min = session.query(func.min(User.age)).one()
    print("最小值: {}".format(age_min))

    # age 求和 只求前三個(gè)的和
    age_sum = session.query(func.sum(User.age)).one()
    print("求總數(shù): {}".format(age_sum))

    # 提交事務(wù)
    session.commit()

from sqlalchemy.orm import sessionmaker
from sqlalchemy.orm import relationship
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import create_engine, Column, Integer, String, Text, func, ForeignKey

# 打開(kāi)數(shù)據(jù)庫(kù)
Base = declarative_base()
engine = create_engine('sqlite:///:memory:', echo=False)

# 主表
class Author(Base):
    __tablename__ = 'author'
    id = Column(Integer, primary_key=True, autoincrement=True)
    name = Column(String(10), nullable=False)

    # 定義外鍵關(guān)聯(lián)到Book模型上面,主表是author
    books = relationship('Book', backref='author')

    def __repr__(self):
        return '<Author:(id={}, name={})>'.format(self.id, self.name)

# 從表
class Book(Base):
    __tablename__ = 'book'
    id = Column(Integer, primary_key=True, autoincrement=True)
    name = Column(String(20), nullable=False)

    # 外鍵關(guān)聯(lián)到主表author的id字段上
    author_id = Column(Integer, ForeignKey('author.id',ondelete="RESTRICT"))

    def __repr__(self):
        return '<Book:(id={}, name={}, author_id={})>'.format(self.id, self.name, self.author_id)

if __name__ == "__main__":
    # 創(chuàng)建會(huì)話
    Session = sessionmaker(bind=engine)
    session = Session()

    # 創(chuàng)建數(shù)據(jù)表(存在則跳過(guò))
    Base.metadata.create_all(engine, checkfirst=True)

    # ------------------------------------------------------
    # 創(chuàng)建數(shù)據(jù)并插入
    author1 = Author(id=1, name="張三")
    author2 = Author(id=2, name="李四")

    book1 = Book(name="<Python 開(kāi)發(fā)>", author_id=1)
    book2 = Book(name="<C++ 開(kāi)發(fā)教程>", author_id=1)
    book3 = Book(name="<C# 從入門(mén)到精通>", author_id=1)

    book4 = Book(name="<滲透測(cè)試指南>", author_id=2)
    book5 = Book(name="<nmap 掃描工具指南>", author_id=2)

    session.add_all([author1,book1,book2,book3])
    session.add_all([author2,book4,book5])
    session.commit()

    # ------------------------------------------------------
    # 關(guān)聯(lián)插入模式

    author1 = Author(id=3, name="王五")

    book1 = Book(name="<Python 開(kāi)發(fā)>", author_id=1)
    book2 = Book(name="<C++ 開(kāi)發(fā)教程>", author_id=1)
    book3 = Book(name="<C# 從入門(mén)到精通>", author_id=1)

    author1.books.append(book1)
    author1.books.append(book2)
    author1.books.append(book3)

    session.add(author1)
    session.commit()

    # ------------------------------------------------------
    # 一對(duì)多查詢測(cè)試
    book = session.query(Book).get(1)
    print("書(shū)籍作者: {}".format(book.author.name))

    author = session.query(Author).get(1)
    print("書(shū)籍?dāng)?shù)量: {}".format(len(author.books)))

    for book_name in author.books:
        print("書(shū)籍: {}".format(book_name.name))

from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker,relationship,backref
from sqlalchemy import create_engine, Column, Integer, String, Text, func, ForeignKey

# 打開(kāi)數(shù)據(jù)庫(kù)
Base = declarative_base()
engine = create_engine('sqlite:///:memory:', echo=False)

# 主表
class User(Base):
    __tablename__ = 'user'
    id = Column(Integer, primary_key=True, autoincrement=True)
    username = Column(String(50), nullable=False)

    def __repr__(self):
        return 'User(username:%s)' % self.username

# 從表
class UserExtend(Base):
    __tablename__ = 'user_extend'
    id = Column(Integer,primary_key=True,autoincrement=True)
    school = Column(String(50))
    age = Column(String(32))
    sex = Column(String(32))

    uid = Column(Integer,ForeignKey('user.id'))
    user = relationship('User',backref=backref('extend', uselist=False))

    #uselist=False 告訴父模型 以后引用時(shí)不再是列表 而是對(duì)象
    def __repr__(self):
        return 'extend(school:%s)'%self.school

if __name__ == "__main__":
    # 創(chuàng)建會(huì)話
    Session = sessionmaker(bind=engine)
    session = Session()

    # 創(chuàng)建數(shù)據(jù)表(存在則跳過(guò))
    Base.metadata.create_all(engine, checkfirst=True)

    # ------------------------------------------------------
    # 插入測(cè)試數(shù)據(jù)
    user = User(username="lyshark")
    extend = UserExtend(school="<家里蹲大學(xué)>", age="22", sex="M")
    extend.user = user

    session.add(extend)
    session.commit()

    # ------------------------------------------------------
    # 一對(duì)一關(guān)系測(cè)試
    user_ptr = session.query(User).first()
    print("用戶名: {} --> 學(xué)校: {}".format(user_ptr.username,user_ptr.extend.school))

    extend_ptr = session.query(UserExtend).first()
    print("用戶名: {} --> 學(xué)校: {} --> 年齡: {} --> 性別: {}".format(extend_ptr.user.username,extend_ptr.school,extend_ptr.age,extend_ptr.sex))

from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker,relationship,backref
from sqlalchemy import create_engine, Column, Integer, String, Text, func, ForeignKey, Table
from sqlalchemy import Column,INT,VARCHAR,ForeignKey
from sqlalchemy.orm import relationship

# 打開(kāi)數(shù)據(jù)庫(kù)
Base = declarative_base()
engine = create_engine('sqlite:///:memory:', echo=False)

# 女孩表
class Girls(Base):
    __tablename__ = "girl"
    id = Column(Integer,primary_key=True, autoincrement=True)
    name = Column(String(32))

    # 建立多對(duì)多關(guān)系
    g2b = relationship("Boys",backref="b2g",secondary="hotel")

# 男孩表
class Boys(Base):
    __tablename__ = "boy"
    id = Column(Integer,primary_key=True, autoincrement=True)
    name = Column(String(32))

# 映射關(guān)系表
class Table(Base):
    __tablename__ = "hotel"
    id = Column(Integer, primary_key=True, autoincrement=True)
    boy_id = Column(Integer,ForeignKey("boy.id"))
    girl_id = Column(Integer,ForeignKey("girl.id"))

if __name__ == "__main__":
    # 創(chuàng)建會(huì)話
    Session = sessionmaker(bind=engine)
    session = Session()

    # 創(chuàng)建數(shù)據(jù)表(存在則跳過(guò))
    Base.metadata.create_all(engine, checkfirst=True)

    # ------------------------------------------------------
    # 增加數(shù)據(jù) - relationship 正向
    girl_obj = Girls(name="女孩主")
    girl_obj.g2b = [Boys(name="男孩從1"),Boys(name="男孩從2")]
    session.add(girl_obj)
    session.commit()

    # 增加數(shù)據(jù) - relationship 反向
    boy_obj = Boys(name="男孩主")
    boy_obj.b2g = [Girls(name="女孩從1"),Girls(name="女孩從2")]
    session.add(boy_obj)
    session.commit()

    # ------------------------------------------------------
    # 正向查詢
    girl_obj_list = session.query(Girls).all()
    for girl_obj in girl_obj_list:
        for boy in girl_obj.g2b:
            print(girl_obj.name,boy.name)

    # 反向查詢
    boy_obj_list = session.query(Boys).all()
    for boy in boy_obj_list:
        for girl in boy.b2g:
            print(girl.name,boy.name)

from sqlalchemy.orm import sessionmaker
from sqlalchemy.orm import relationship
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import create_engine, Column, Integer, String, Text, func, ForeignKey

# 打開(kāi)數(shù)據(jù)庫(kù)
Base = declarative_base()
engine = create_engine('sqlite:///:memory:', echo=False)

# 主表
class Author(Base):
    __tablename__ = 'author'
    id = Column(Integer, primary_key=True, autoincrement=True)
    name = Column(String(10), nullable=False)

    # 定義外鍵關(guān)聯(lián)到Book模型上面,主表是author
    books = relationship('Book', backref='author')

    def __repr__(self):
        return '<Author:(id={}, name={})>'.format(self.id, self.name)

# 從表
class Book(Base):
    __tablename__ = 'book'
    id = Column(Integer, primary_key=True, autoincrement=True)
    name = Column(String(20), nullable=False)

    # 外鍵關(guān)聯(lián)到主表author的id字段上
    author_id = Column(Integer, ForeignKey('author.id',ondelete="RESTRICT"))

    def __repr__(self):
        return '<Book:(id={}, name={}, author_id={})>'.format(self.id, self.name, self.author_id)

if __name__ == "__main__":
    # 創(chuàng)建會(huì)話
    Session = sessionmaker(bind=engine)
    session = Session()

    # 創(chuàng)建數(shù)據(jù)表(存在則跳過(guò))
    Base.metadata.create_all(engine, checkfirst=True)

    # ------------------------------------------------------
    # 創(chuàng)建數(shù)據(jù)并插入
    author1 = Author(id=1, name="張三")
    author2 = Author(id=2, name="李四")

    book1 = Book(name="<Python 開(kāi)發(fā)>", author_id=1)
    book2 = Book(name="<C++ 開(kāi)發(fā)教程>", author_id=1)
    book3 = Book(name="<C# 從入門(mén)到精通>", author_id=1)

    book4 = Book(name="<滲透測(cè)試指南>", author_id=2)
    book5 = Book(name="<nmap 掃描工具指南>", author_id=2)

    session.add_all([author1,book1,book2,book3])
    session.add_all([author2,book4,book5])
    session.commit()

    # ------------------------------------------------------
    # 連表查詢
    no_join_select = session.query(Author).filter(Author.id == Book.id).filter(Author.name == "王五").all()
    print("查詢主鍵==從鍵 并且 Author.name == 王五的記錄: {}".format(no_join_select))

    # JOIN 連接查詢
    join = session.query(Author).join(Book).filter(Book.name=="<nmap 掃描工具指南>").first().name
    print("查詢主表Author中的Book書(shū)名的作者是: {}".format(join))

    join = session.query(Book).join(Author).filter(Author.name=="李四").all()
    for book in join:
        print("查詢從表Book中的Author作者有哪些書(shū): {}".format(book.name))

    # subquery 子查詢
    sbq = session.query(Book.author_id,func.count('*').label("book_count")).group_by(Book.author_id).subquery()
    print("查詢出書(shū)籍編號(hào)計(jì)數(shù)(子語(yǔ)句): {}".format(sbq))

    sub_join = session.query(Author.name,sbq.c.book_count).outerjoin(sbq,Author.id == sbq.c.author_id).all()
    print("查詢用戶有幾本書(shū)(主語(yǔ)句): {}".format(sub_join))