當前位置：首頁 > 运维知识 > 数据库 >内容正文

数据库

SQLAlchemy 教程 —— 基础入门篇

發布時間：2025/3/15 数据库 19 豆豆

生活随笔收集整理的這篇文章主要介紹了 SQLAlchemy 教程 —— 基础入门篇小編覺得挺不錯的,現在分享給大家,幫大家做個參考.

參照博客:https://blog.csdn.net/qq_43355223/article/details/83024430

filter operator :

檢索返回的列表，以及列表的標量：

一篇好的初探文章：https://www.cnblogs.com/franknihao/p/7268752.html

? ? ? ?session.query(xxx)得到的是一個query對象，上面說過很多后面可以接上的方法，實際上還有更多如下：

　　one()　　如果返回行數不為1，那么就報錯；若剛好返回結果就一條就返回這條記錄的對象

　　limit(n)　　最多只返回n條結果

　　offset(n)　　直接跳過前n條記錄，從n+1條開始返回

　　order_by(Table.attribute 或者 'attribute')　　返回結果按照給出的字段排序。

　　order_by(User.name.desc()) 或者 order_by('name desc')

　　filter(condition1).filter(condition2)　　多個拼接的filter就相當于and_(condition1,condition2...)

　　請注意以上所有方法都要在all()之前調用，all出來已經是一個列表了，不能在作為調用方調用這些方法了。

　　還可以些數據庫自帶的函數，在用之前記得from sqlalchemy import func，就可以通過func來調用了。這些函數不是放在調用鏈中，大多數時候都是放在query方法的參數位置，

　　比如func.count()　　統計一共有多少條記錄作為結果等等

計數：
可以直接使用 .count()計算有多少行：
session.query(User).filter(User.name.like('%ed')).count()

當需要將指定的東西進行計數，就使用func.count()：
session.query(func.count(User.id)).scalar()

使用query.join()結合兩張表來查詢數據表數據：

result = session.query(AlchemyTest).join(AlchemyTestError).filter(AlchemyTestError.errorname == 'cuowu').first() #查詢AlchemyTest表內的數據，看哪些user有這個error if result is not None:print(result.TestName)

因為這兩張表只有一個外鍵，所以query.join()知道怎么關聯這兩張表，但是如果沒有或者不止一個外鍵的時候，使用其他的形式：
result = session.query(AlchemyTest).join((AlchemyTestError, AlchemyTest.TestName==AlchemyTestError.TestName)).first()

這里有一個join的reminder：

outerjoin:?query.outerjoin(User.addresses) # LEFT OUTER JOIN

一、課程簡介

1.1 實驗內容

本課程帶領大家使用 SQLAlchemy 連接 MySQL 數據庫，創建一個博客應用所需要的數據表，并介紹了使用 SQLAlchemy 進行簡單了 CURD 操作及使用 Faker 生成測試數據。

1.2課程知識點

學會用 SQLALchemy 連接數據庫(MySQL, SQLite, PostgreSQL), 創建數據表；
掌握表數據之間一對一，一對多及多對多的關系并能轉化為對應 SQLAlchemy 描述；
掌握使用 SQLAlchemy 進行 CURD 操作；
學會使用 Faker 生成測試數據

學習本課程需要你對 Python 與 MySQL 都有基本的掌握。

二、ORM 與 SQLAlchemy 簡介

ORM 全稱?Object Relational Mapping, 翻譯過來叫對象關系映射。簡單的說，ORM 將數據庫中的表與面向對象語言中的類建立了一種對應關系。這樣，我們要操作數據庫，數據庫中的表或者表中的一條記錄就可以直接通過操作類或者類實例來完成。

SQLAlchemy 是Python 社區最知名的 ORM 工具之一，為高效和高性能的數據庫訪問設計，實現了完整的企業級持久模型。

接下來我們將使用 SQLAlchemy 和 MySQL 構建一個博客應用的實驗庫。

先安裝 SQLAlchemy:

sudo pip install sqlalchemy

三、連接與創建

實驗樓環境已經為我們安裝了 MySQL，但還沒有啟動, 在啟動 MySQL 之前，我們需要進行一些配置，將 MySQL 默認的 latin1 編碼改成 utf8 。

sudo vim /etc/mysql/my.cnf

通過上面的命令打開 MySQL 的配置文件，添加下面幾個配置：

[client] default-character-set = utf8[mysqld] character-set-server = utf8[mysql] default-character-set = utf8

保存退出。現在我們可以啟動 MySQL 服務了：

sudo service mysql start

在命令行下輸入下面命令啟動 MySQL：

mysql -uroot -p

看到上面的內容就說明我們的 MySQL 可以正常啟動了（注意，上面的密碼不需要輸入內容，直接回車就行）, 并且我們我們通過命令：

> create database blog;

創建一個名為?blog?的數據庫為下面的使用作準備。

另外，我們需要安裝一個 Python 與 MySQL 之間的驅動程序：

sudo apt-get install python-mysqldb

3.1 連接數據庫

我們在 Code 下新建個 Python 文件，叫什么名字就隨你便了，這里我們叫?db.py，寫入下面的內容：

# coding: utf-8from sqlalchemy import create_engine engine = create_engine('mysql+mysqldb://root@localhost:3306/blog') print(engine)

在上面的程序中，我們連接了默認運行在?3306?端口的 MySQL 中的?blog?數據庫。

運行下這個程序，看到下面的信息說明我們已經連接成功了：

3.2 描述表結構

要使用 ORM, 我們需要將數據表的結構用 ORM 的語言描述出來。SQLAlchmey 提供了一套 Declarative 系統來完成這個任務。我們以創建一個?users?表為例，看看它是怎么用 SQLAlchemy 的語言來描述的：

# coding: utf-8from sqlalchemy import create_engine from sqlalchemy.ext.declarative import declarative_base from sqlalchemy import Column, String, Integerengine = create_engine('mysql+mysqldb://root@localhost:3306/blog?charset=utf8') Base = declarative_base()class User(Base):__tablename__ = 'users'id = Column(Integer, primary_key=True)username = Column(String(64), nullable=False, index=True)password = Column(String(64), nullable=False)email = Column(String(64), nullable=False, index=True)def __repr__(self):return '%s(%r)' % (self.__class__.__name__, self.username)

我們看到，在?User?類中，用?__tablename__?指定在 MySQL 中表的名字。我們創建了三個基本字段，類中的每一個?Column?代表數據庫中的一列，在?Colunm中，指定該列的一些配置。第一個字段代表類的數據類型，上面我們使用?String,?Integer?倆個最常用的類型，其他常用的包括：

Text Boolean SmallInteger DateTime

nullable=False?代表這一列不可以為空，index=True?表示在該列創建索引。

另外定義?__repr__?是為了方便調試，你可以不定義，也可以定義的更詳細一些。

python db.py

運行程序，我們在 MySQL 中看看表是如何創建的：

四、關系定義

4.1 一對多關系

對于一個普通的博客應用來說，用戶和文章顯然是一個一對多的關系，一篇文章屬于一個用戶，一個用戶可以寫很多篇文章，那么他們之間的關系可以這樣定義：

class User(Base):__tablename__ = 'users'id = Column(Integer, primary_key=True)username = Column(String(64), nullable=False, index=True)password = Column(String(64), nullable=False)email = Column(String(64), nullable=False, index=True)articles = relationship('Article')def __repr__(self):return '%s(%r)' % (self.__class__.__name__, self.username)class Article(Base):__tablename__ = 'articles'id = Column(Integer, primary_key=True)title = Column(String(255), nullable=False, index=True)content = Column(Text)user_id = Column(Integer, ForeignKey('users.id'))author = relationship('User')def __repr__(self):return '%s(%r)' % (self.__class__.__name__, self.title)

每篇文章有一個外鍵指向?users?表中的主鍵?id，而在?User?中使用 SQLAlchemy 提供的?relationship?描述關系。而用戶與文章的之間的這個關系是雙向的，所以我們看到上面的兩張表中都定義了?relationship。

SQLAlchemy 提供了?backref?讓我們可以只需要定義一個關系：

articles = relationship('Article', backref='author')

添加了這個就可以不用再在?Article?中定義?relationship?了！

4.2 一對一關系

在?User?中我們只定義了幾個必須的字段，但通常用戶還有很多其他信息，但這些信息可能不是必須填寫的，我們可以把它們放到另一張?UserInfo?表中，這樣User?和?UserInfo?就形成了一對一的關系。你可能會奇怪一對一關系為什么不在一對多關系前面？那是因為一對一關系是基于一對多定義的：

class User(Base):__tablename__ = 'users'id = Column(Integer, primary_key=True)username = Column(String(64), nullable=False, index=True)password = Column(String(64), nullable=False)email = Column(String(64), nullable=False, index=True)articles = relationship('Article', backref='author')userinfo = relationship('UserInfo', backref='user', uselist=False)def __repr__(self):return '%s(%r)' % (self.__class__.__name__, self.username)class UserInfo(Base):__tablename__ = 'userinfos'id = Column(Integer, primary_key=True)name = Column(String(64))qq = Column(String(11))phone = Column(String(11))link = Column(String(64))user_id = Column(Integer, ForeignKey('users.id'))

定義方法和一對多相同，只是需要添加?userlist=False?。

4.3 多對多關系

一遍博客通常有一個分類，好幾個標簽。標簽與博客之間就是一個多對多的關系。多對多關系不能直接定義，需要分解成倆個一對多的關系，為此，需要一張額外的表來協助完成：

article_tag = Table('article_tag', Base.metadata,Column('article_id', Integer, ForeignKey('articles.id')),Column('tag_id', Integer, ForeignKey('tags.id')) )class Tag(Base):__tablename__ = 'tags'id = Column(Integer, primary_key=True)name = Column(String(64), nullable=False, index=True)def __repr__(self):return '%s(%r)' % (self.__class__.__name__, self.name)

4.4 映射到數據

表已經描述好了，在文件末尾使用下面的命令在我們連接的數據庫中創建對應的表：

if __name__ == '__main__':Base.metadata.create_all(engine)

進入 MySQL 看看：

所有的表都已經創建好了！

五、簡單 CURD

當你想打電話給朋友時，你是否得用手機撥通他的號碼才能建立起一個會話？同樣的，你想和 MySQL 交談也得先通過 SQLAlchemy 建立一個會話：

from sqlalchemy.orm import sessionmakerSession = sessionmaker(bind=engine) session = Session()

你可以把?sessionmaker?想象成一個手機，engine?當做 MySQL 的號碼，撥通這個“號碼”我們就創建了一個 Session 類，下面就可以通過這個類的實例與 MySQL 愉快的交談了！

5.1 Create

如果你玩過LOL, 我想你一定知道Faker。而在 Python的世界中，Faker 是用來生成虛假數據的庫。安裝它：

sudo pip install faker

下面結合 Faker 庫創建一些測試數據：

faker = Factory.create()Session = sessionmaker(bind=engine)session = Session()faker_users = [User(username=faker.name(),password=faker.word(),email=faker.email(),) for i in range(10)]session.add_all(faker_users)faker_categories = [Category(name=faker.word()) for i in range(5)]session.add_all(faker_categories)faker_tags= [Tag(name=faker.word()) for i in range(20)]session.add_all(faker_tags)for i in range(100):article = Article(title=faker.sentence(),content=' '.join(faker.sentences(nb=random.randint(10, 20))),author=random.choice(faker_users),category=random.choice(faker_categories))for tag in random.sample(faker_tags, random.randint(2, 5)):article.tags.append(tag)session.add(article)session.commit()

在上面的代碼中我們創建了10個用戶，5個分類，20個標簽，100篇文章，并且為每篇文章隨機選擇了2~5個標簽。

使用 SQLAlchemy 往數據庫中添加數據，我們只需要創建相關類的實例，調用?session.add()?添加一個，或者?session.add_all()?一次添加多個，最后?session.commit()?就可以了。

5.3 Update

更新一個字段：

>>> a = session.query(Article).get(10) >>> a.title = 'My test blog post' >>> session.add(a) >>> session.commit()

添加一個標簽：

>>> a = session.query(Article).get(10) >>> a.tags.append(Tag(name='python')) >>> session.add(a) >>> session.commit()

5.4 Delete

>>> a = session.query(Article).get(10) >>> session.delete(a) >>> session.commit()

刪除直接調用?delete?刪除獲取到的對象，提交 session 即可。

完整代碼

# coding: utf-8import random from faker import Factoryfrom sqlalchemy import create_engine, Table from sqlalchemy.ext.declarative import declarative_base #提取聲明 from sqlalchemy import ForeignKey #外鍵 from sqlalchemy import Column, String, Integer, Text from sqlalchemy.orm import sessionmaker, relationship #創建會議、關系engine = create_engine('mysql+mysqldb://root@localhost:3306/blog?charset=utf8') Base = declarative_base()class User(Base):__tablename__ = 'users'id = Column(Integer, primary_key=True)username = Column(String(64), nullable=False, index=True)password = Column(String(64), nullable=False)email = Column(String(64), nullable=False, index=True)articles = relationship('Article', backref='author')userinfo = relationship('UserInfo', backref='user', uselist=False)def __repr__(self):return '%s(%r)' % (self.__class__.__name__, self.username)class UserInfo(Base):__tablename__ = 'userinfos'id = Column(Integer, primary_key=True)name = Column(String(64))qq = Column(String(11))phone = Column(String(11))link = Column(String(64))user_id = Column(Integer, ForeignKey('users.id'))class Article(Base):__tablename__ = 'articles'id = Column(Integer, primary_key=True)title = Column(String(255), nullable=False, index=True)content = Column(Text)user_id = Column(Integer, ForeignKey('users.id'))cate_id = Column(Integer, ForeignKey('categories.id'))tags = relationship('Tag', secondary='article_tag', backref='articles')def __repr__(self):return '%s(%r)' % (self.__class__.__name__, self.title)class Category(Base):__tablename__ = 'categories'id = Column(Integer, primary_key=True)name = Column(String(64), nullable=False, index=True)articles = relationship('Article', backref='category')def __repr__(self):return '%s(%r)' % (self.__class__.__name__, self.name)article_tag = Table('article_tag', Base.metadata,Column('article_id', Integer, ForeignKey('articles.id')),Column('tag_id', Integer, ForeignKey('tags.id')) )class Tag(Base):__tablename__ = 'tags'id = Column(Integer, primary_key=True)name = Column(String(64), nullable=False, index=True)def __repr__(self):return '%s(%r)' % (self.__class__.__name__, self.name)if __name__ == '__main__':Base.metadata.create_all(engine)faker = Factory.create()Session = sessionmaker(bind=engine)session = Session()faker_users = [User(username=faker.name(),password=faker.word(),email=faker.email(),) for i in range(10)]session.add_all(faker_users)faker_categories = [Category(name=faker.word()) for i in range(5)]session.add_all(faker_categories)faker_tags= [Tag(name=faker.word()) for i in range(20)]session.add_all(faker_tags)for i in range(100):article = Article(title=faker.sentence(),content=' '.join(faker.sentences(nb=random.randint(10, 20))),author=random.choice(faker_users),category=random.choice(faker_categories))for tag in random.sample(faker_tags, random.randint(2, 5)):article.tags.append(tag)session.add(article)session.commit()

六、總結與習題

本篇教程帶領大家使用 SQLAlchemy 創建了一個博客應用的數據庫及相關表，再次過程中，介紹了使用 SQLAlchemy 定義一對一、一對多及多對多關系，CURD 及使用 Faker 生成測試數據。

總結

以上是生活随笔為你收集整理的SQLAlchemy 教程 —— 基础入门篇的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。

上一篇： Nmap源码分析（整体架构）
下一篇： twisted系列教程十二–为serve