使用 pip 安裝
# 資料庫
pip install pymysql sqlalchemy
下載資料庫管理軟體 HeidiSQL 並安裝
https://www.heidisql.com/download.php
使用 Docker 快速建立 MySQL 伺服器
https://hub.docker.com/_/mariadb
docker pull mariadb:10.4.7
# 建立一個帳號密碼均為 root,且有一個名為 trader 資料庫的 MySQL 伺服器
docker run -d -p 3306:3306 -e MYSQL_ROOT_PASSWORD=root -e MYSQL_DATABASE=trader --name mariadb mariadb:10.4.7
docker ps -a
容器狀態應有 UP 字串,代表容器正在運行。
為了降低困難度,使用 SQLAlchemy ORM 來簡化操作。
USE test;
CREATE TABLE `stocks` (
`id` INT(11) NOT NULL AUTO_INCREMENT,
`code` CHAR(4) NOT NULL COLLATE 'utf8mb4_unicode_ci',
`name` VARCHAR(32) NOT NULL COLLATE 'utf8mb4_unicode_ci',
`created_at` TIMESTAMP NOT NULL DEFAULT current_timestamp(),
`updated_at` TIMESTAMP NOT NULL DEFAULT current_timestamp() ON UPDATE current_timestamp(),
`deleted_at` TIMESTAMP NULL DEFAULT NULL,
PRIMARY KEY (`id`)
)
COLLATE='utf8mb4_unicode_ci'
ENGINE=InnoDB;
import datetime
import loguru
import sqlalchemy
import sqlalchemy.ext.automap
import sqlalchemy.orm
import sqlalchemy.schema
def main():
username = 'root' # 資料庫帳號
password = 'root' # 資料庫密碼
host = 'localhost' # 資料庫位址
port = '3306' # 資料庫埠號
database = 'trader' # 資料庫名稱
# 建立連線引擎
engine = sqlalchemy.create_engine(
f'mysql+pymysql://{username}:{password}@{host}:{port}/{database}'
)
# 取得資料庫元資料
metadata = sqlalchemy.schema.MetaData(engine)
# 產生自動對應參照
automap = sqlalchemy.ext.automap.automap_base()
automap.prepare(engine, reflect=True)
# 準備 ORM 連線
session = sqlalchemy.orm.Session(engine)
# 載入 stocks 資料表資訊
sqlalchemy.Table('stocks', metadata, autoload=True)
# 取出對應 stocks 資料表的類別
Stock = automap.classes['stocks']
try:
# 執行原生 SQL 命令清空資料庫
session.execute('TRUNCATE TABLE stocks')
# 送出執行命令
session.commit()
except Exception as e:
# 發生例外錯誤,還原交易
session.rollback()
loguru.logger.error('清空資料失敗')
# 進入交易模式
try:
# 建立第一筆資料
stock = Stock()
stock.code = '1101'
stock.name = '台泥'
session.add(stock)
# 建立第二筆資料
stock = Stock()
stock.code = '1102'
stock.name = '亞泥'
session.add(stock)
# 建立第三筆資料
stock = Stock()
stock.code = '1103'
stock.name = '嘉泥'
session.add(stock)
# 建立第四筆資料
stock = Stock()
stock.code = '1201'
stock.name = '味全'
session.add(stock)
# 寫入資料庫
session.commit()
except Exception as e:
# 發生例外錯誤,還原交易
session.rollback()
loguru.logger.error('新增資料失敗')
loguru.logger.error(e)
# 關閉連線
session.close()
if __name__ == '__main__':
loguru.logger.add(
f'{datetime.date.today():%Y%m%d}.log',
rotation='1 day',
retention='7 days',
level='DEBUG'
)
main()
import datetime
import loguru
import sqlalchemy
import sqlalchemy.ext.automap
import sqlalchemy.orm
import sqlalchemy.schema
def main():
username = 'root' # 資料庫帳號
password = 'root' # 資料庫密碼
host = 'localhost' # 資料庫位址
port = '3306' # 資料庫埠號
database = 'trader' # 資料庫名稱
# 建立連線引擎
engine = sqlalchemy.create_engine(
f'mysql+pymysql://{username}:{password}@{host}:{port}/{database}'
)
metadata = sqlalchemy.schema.MetaData(engine)
automap = sqlalchemy.ext.automap.automap_base()
automap.prepare(engine, reflect=True)
session = sqlalchemy.orm.Session(engine)
sqlalchemy.Table('stocks', metadata, autoload=True)
Stock = automap.classes['stocks']
loguru.logger.info('取出資料表所有資料')
results = session.query(Stock).all()
for stock in results:
loguru.logger.info(f'{stock.code} {stock.name}')
loguru.logger.info('使用過濾條件取出資料表資料')
results = session.query(Stock).filter(
Stock.code == '1102'
).all()
for stock in results:
loguru.logger.info(f'{stock.code} {stock.name}')
loguru.logger.info('使用多重過濾條件 OR 取出資料表資料')
results = session.query(Stock).filter(
sqlalchemy.or_(
Stock.code.like('%02'),
Stock.code.like('12%')
)
).all()
for stock in results:
loguru.logger.info(f'{stock.code} {stock.name}')
loguru.logger.info('使用多重過濾條件 AND 取出資料表資料')
results = session.query(Stock).filter(
sqlalchemy.and_(
Stock.code.like('%2'),
Stock.name.like('%泥')
)
).all()
for stock in results:
loguru.logger.info(f'{stock.code} {stock.name}')
loguru.logger.info('使用排序取出資料表資料')
results = session.query(Stock).order_by(Stock.code.desc()).all()
for stock in results:
loguru.logger.info(f'{stock.code} {stock.name}')
session.close()
if __name__ == '__main__':
loguru.logger.add(
f'{datetime.date.today():%Y%m%d}.log',
rotation='1 day',
retention='7 days',
level='DEBUG'
)
main()
import datetime
import loguru
import sqlalchemy
import sqlalchemy.ext.automap
import sqlalchemy.orm
import sqlalchemy.schema
def main():
username = 'root' # 資料庫帳號
password = 'root' # 資料庫密碼
host = 'localhost' # 資料庫位址
port = '3306' # 資料庫埠號
database = 'trader' # 資料庫名稱
# 建立連線引擎
engine = sqlalchemy.create_engine(
f'mysql+pymysql://{username}:{password}@{host}:{port}/{database}'
)
metadata = sqlalchemy.schema.MetaData(engine)
automap = sqlalchemy.ext.automap.automap_base()
automap.prepare(engine, reflect=True)
session = sqlalchemy.orm.Session(engine)
sqlalchemy.Table('stocks', metadata, autoload=True)
Stock = automap.classes['stocks']
loguru.logger.info('----- 更新單筆資料 -----')
try:
stock = session.query(Stock).filter(
Stock.code == '1102'
).one()
stock.name = stock.name + '(水泥產業)'
session.add(stock)
# 寫入資料庫
session.commit()
except Exception as e:
# 發生例外錯誤,還原交易
session.rollback()
loguru.logger.error('更新資料失敗')
loguru.logger.error(e)
loguru.logger.info('取出資料表所有資料')
results = session.query(Stock).all()
for stock in results:
loguru.logger.info(f'{stock.code} {stock.name}')
loguru.logger.info('----- 更新多筆資料 -----')
try:
session.query(Stock).filter(
sqlalchemy.and_(
Stock.code.like('11%'),
Stock.code != '1102'
)
).update({
Stock.name: Stock.name + '(水泥產業)'
}, synchronize_session=False)
# 寫入資料庫
session.commit()
except Exception as e:
# 發生例外錯誤,還原交易
session.rollback()
loguru.logger.error('更新資料失敗')
loguru.logger.error(e)
loguru.logger.info('取出資料表所有資料')
results = session.query(Stock).all()
for stock in results:
loguru.logger.info(f'{stock.code} {stock.name}')
session.close()
if __name__ == '__main__':
loguru.logger.add(
f'{datetime.date.today():%Y%m%d}.log',
rotation='1 day',
retention='7 days',
level='DEBUG'
)
main()
import datetime
import loguru
import sqlalchemy
import sqlalchemy.ext.automap
import sqlalchemy.orm
import sqlalchemy.schema
def main():
username = 'root' # 資料庫帳號
password = 'root' # 資料庫密碼
host = 'localhost' # 資料庫位址
port = '3306' # 資料庫埠號
database = 'trader' # 資料庫名稱
# 建立連線引擎
engine = sqlalchemy.create_engine(
f'mysql+pymysql://{username}:{password}@{host}:{port}/{database}'
)
metadata = sqlalchemy.schema.MetaData(engine)
automap = sqlalchemy.ext.automap.automap_base()
automap.prepare(engine, reflect=True)
session = sqlalchemy.orm.Session(engine)
sqlalchemy.Table('stocks', metadata, autoload=True)
Stock = automap.classes['stocks']
loguru.logger.info('----- 刪除單筆資料 -----')
try:
stock = session.query(Stock).filter(
Stock.code == '1102'
).one()
session.delete(stock)
# 寫入資料庫
session.commit()
except Exception as e:
# 發生例外錯誤,還原交易
session.rollback()
loguru.logger.error('刪除資料失敗')
loguru.logger.error(e)
loguru.logger.info('取出資料表所有資料')
results = session.query(Stock).all()
for stock in results:
loguru.logger.info(f'{stock.code} {stock.name}')
loguru.logger.info('----- 刪除多筆資料 -----')
try:
session.query(Stock).filter(
Stock.code.like('11%')
).delete(synchronize_session=False)
# 寫入資料庫
session.commit()
except Exception as e:
# 發生例外錯誤,還原交易
session.rollback()
loguru.logger.error('刪除資料失敗')
loguru.logger.error(e)
loguru.logger.info('取出資料表所有資料')
results = session.query(Stock).all()
for stock in results:
loguru.logger.info(f'{stock.code} {stock.name}')
session.close()
if __name__ == '__main__':
loguru.logger.add(
f'{datetime.date.today():%Y%m%d}.log',
rotation='1 day',
retention='7 days',
level='DEBUG'
)
main()
https://pymysql.readthedocs.io/
https://docs.sqlalchemy.org/
團隊系列文:
CSScoke - 金魚都能懂的這個網頁畫面怎麼切 - 金魚都能懂了你還怕學不會嗎
Clarence - LINE bot 好好玩 30 天玩轉 LINE API
Hina Hina - 陣列大亂鬥
King Tzeng - IoT沒那麼難!新手用JavaScript入門做自己的玩具
Vita Ora - 好 Js 不學嗎 !? JavaScript 入門中的入門。
TaTaMo - 用Python開發的網頁不能放到Github上?Lektor說可以!!