# 方法 1: SQLAlchemy 2.0 推荐的方式
from sqlalchemy.orm import DeclarativeBase
class Base(DeclarativeBase):
pass
# 方法 2
from sqlalchemy.orm import declarative_base
Base = declarative_base()
Base.metadata # sqlalchemy.sql.schema.MetaData 对象, 与 Core API 里显式用 metadata_obj = MetaData() 得到的 metadata_obj 对应
Step 2: 继承 Base
from typing import List
from typing import Optional
from sqlalchemy.orm import Mapped
from sqlalchemy.orm import mapped_column
from sqlalchemy import Column
from sqlalchemy.orm import relationship
class User(Base):
__tablename__ = "user_account" # 数据库里的表名
id: Mapped[int] = mapped_column(primary_key=True)
# id = Column(Integer, primary_key=True) # 这种写法是类似于 Core API 的写法, 许多网上的资料会用这种写法
name: Mapped[str] = mapped_column(String(30))
fullname: Mapped[Optional[str]]
addresses: Mapped[List["Address"]] = relationship(back_populates="user") # 不存储在数据库里
def __repr__(self) -> str:
return f"User(id={self.id!r}, name={self.name!r}, fullname={self.fullname!r})"
class Address(Base):
__tablename__ = "address"
id: Mapped[int] = mapped_column(primary_key=True)
email_address: Mapped[str]
user_id = mapped_column(ForeignKey("user_account.id")) # 定义外键约束, 数据库层面的约束
user: Mapped[User] = relationship(back_populates="addresses")
def __repr__(self) -> str:
return f"Address(id={self.id!r}, email_address={self.email_address!r}, user_id={self.user_id!r})"
relationship 是可选的, 实际的数据库存储里并不包含 user 及 addresses 这两列, 它与 ForeignKey 的关系以及给 ORM API 带来的便利性具体见后续, 大体如下(TODO: 待确认)
# 假设 user 是一个 User 对象, 在 User 里定义了 addresses = relationship(back_populates="user")
user.addresses[0].email_address
from sqlalchemy.orm import Session
with Session(engine) as session:
sandy = session.query(User).filter_by(name="sandy").first()
if sandy:
for addr in sandy.addresses:
print(addr.email_address)
from sqlalchemy import create_engine
from sqlalchemy import select, text
from sqlalchemy.orm import sessionmaker
from sql_test import Address, User
engine = create_engine('sqlite:///example.db', echo=False)
Session = sessionmaker(bind=engine)
with Session() as session:
result = session.execute(text("SELECT * FROM user_account"))
for row in result:
print(type(row), row) # (sqlalchemy.engine.row.Row, (1, 'Ask', 'AskBob'))
result1 = session.execute(select(User))
for row1 in result1:
print(type(row), row1) # (sqlalchemy.engine.row.Row, (User(id=1, name='Ask', fullname='AskBob'),))
# 两种查询方式结果有所不同
# for item in result.mappings(): print(item)
# {'id': 1, 'name': 'Ask', 'fullname': 'AskBob'}
# for item in result1.mappings(): print(item)
# {'User': User(id=1, name='Ask', fullname='AskBob')}
Core
from sqlalchemy import select, text
stmt1 = text("select * from xx") # stmt1 是一个 TextClause 对象, sqlalchemy.sql.elements.TextClause
stmt2 = select(User) # stmt2 是一个 Select 对象 (sqlalchemy.sql.selectable.Select)
# 打印 SQL 语句
print(stmt1) # "select * from xx"
print(stmt2) # "SELECT user_account.id, user_account.name, user_account.fullname FROM user_account"
# 备注: stmt2 转化为 SQL 语句字符串的过程会先经过 compile
# compiled = stmt2.compile()
result = connect.execute(stmt2)
# pip install pymilvus[model]
from pymilvus.model.hybrid import BGEM3EmbeddingFunction
docs = [
"Artificial intelligence was founded as an academic discipline in 1956.",
"Alan Turing was the first person to conduct substantial research in AI.",
"Born in Maida Vale, London, Turing was raised in southern England.",
]
query = "Who started AI research?"
bge_m3_ef = BGEM3EmbeddingFunction(use_fp16=False, device="cpu")
docs_embeddings = bge_m3_ef(docs)
query_embeddings = bge_m3_ef([query])
res = collection.query(expr='title like "The%"', output_fields=["id", "title"])
res = collection.query(expr='title like "%the%"', output_fields=["id", "title"])
res = collection.query(expr='title like "%Rye"', output_fields=["id", "title"])
res = collection.query(expr='title like "Flip_ed"', output_fields=["id", "title"]) # _ 代表一个任意字符
# you can create inverted index to accelerate the fuzzy match.
collection.release()
collection.create_index(
"title", {"index_type": "INVERTED"})
collection.load()