安装

pip install sqlalchemy

创建数据库连接

dialect+driver://username:password@host:port/database

dialect：数据库，如：sqlite、mysql、oracle等
driver：数据库驱动，用于连接数据库的，本文使用pymysql
username：用户名
password：密码
host：IP地址
port：端口
database：数据库

Sqlite

# 相对路径
engine = create_engine('sqlite:///foo.db')

# 绝对路径
#Unix/Mac - 4 initial slashes in total
engine = create_engine('sqlite:////absolute/path/to/foo.db')
#Windows
engine = create_engine('sqlite:///C:\\path\\to\\foo.db')

Mysql

pip install pymysql

HOST = 'localhost'
PORT = 3306
USERNAME = 'root'
PASSWORD = '123456'
DB = 'myclass'

# dialect + driver://username:passwor@host:port/database
DB_URI = f'mysql+pymysql://{USERNAME}:{PASSWORD}@{HOST}:{PORT}/{DB}'

# default
engine = create_engine('mysql://scott:tiger@localhost/foo')
# mysql-python
engine = create_engine('mysql+mysqldb://scott:tiger@localhost/foo')

定义配置项： Config.py

import os

class Config:

    ROOT_PATH = os.path.dirname(__file__)

数据库操纵类 DB.py

from Config import Config
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker


class DB:
    __db_session = None
    __db_engine = None
    __sqlite_file = 'sqlite3.db'

    def get_db_engine(self):
        if self.__db_engine is None:
            if self.__sqlite_file.startswith('/'):
                self.__db_engine = create_engine('sqlite:///' + self.__sqlite_file, echo=True)
            self.__db_engine = create_engine('sqlite:///' + Config.ROOT_PATH + '/' + self.__sqlite_file, echo=True)
        return self.__db_engine

    def get_db_session(self):
        if self.__db_session is None:
            engine = self.get_db_engine()
            self.__db_session = sessionmaker(engine)()
        return self.__db_session

数据模型基类 models/Base.py

from sqlalchemy.ext.declarative import declarative_base, declared_attr
from sqlalchemy import create_engine, Column, Integer, DateTime
import time


class DeclarativeBase(object):

    # @declared_attr
    # def created_at(self):
    #     return Column(Integer, default=time.time())

    id = Column(Integer, primary_key=True)
    created_at = Column(Integer, default=int(time.time()))
    updated_at = Column(Integer, default=int(time.time()))
    # created_at = Column(DateTime, default=time.strftime("%Y-%m-%d %H:%M:%S", time.localtime()))


Base = declarative_base(cls=DeclarativeBase)

模型类 models/User.py

from sqlalchemy import Column, String, Integer
from models.Base import Base
from DB import DB


class User(Base):

    __tablename__ = 'user'
    name = Column(String(32))
    password = Column(String(64))

    @staticmethod
    def create_table(engine):
        User.__table__.create(engine, checkfirst=True)
        # Base.metadata.create_all(engine)

if __name__ == '__main__':
    db = DB()
    # User.create_table(db.get_db_engine())
    session = db.get_db_session()
    one = session.query(User).filter_by(name='Harvey').order_by(User.created_at.desc()).first()
    all = session.query(User).filter_by(name='Harvey').all()
    print(one.password)
    for user in all:
        print(user.password)
    # user = User(name='Tom', password='pwd_tom_5555')
    # session.add(user)
    # session.commit()

Python3+SQLAlchemy+Sqlite3实现ORM教程 https://www.cnblogs.com/lsdb/p/9835894.html
sqlalchemy的基本用法 https://www.cnblogs.com/blueberry-mint/p/14277882.html