当前位置：首页 > news >正文

SQLAlchemy示例（连接数据库插入表数据）

news 2025/7/6 20:53:07

背景需求

连接数据库，插入表中一些数据。

其用户是新建用户，所以只能插入，不能更新。

再次输入数据则使用更新数据语法，这个没调试。

#! /usr/bin/env python
# -*- coding: utf-8 -*-from sqlalchemy import create_engine, Column, BigInteger, Integer, String, SmallInteger, DECIMAL, DateTime
from sqlalchemy.orm import declarative_base, sessionmaker
from datetime import datetime# 1. 创建一个基本的 ORM 映射类
Base = declarative_base()# 2. 创建一个映射到数据库表的类（模型）
class AAAAAA(Base):__tablename__ = 'aaaaaaa'  # 数据库中的表名id = Column(BigInteger, primary_key=True)uid = Column(BigInteger)mid = Column(BigInteger)account_type = Column(SmallInteger)variety_code = Column(String(20))balance = Column(DECIMAL(20, 8))available = Column(DECIMAL(20, 8))freeze = Column(DECIMAL(20, 8))create_time = Column(DateTime)update_time = Column(DateTime)def __repr__(self):# return f"<AAAAAA(uid={self.uid}, mid={self.mid}, account_type={self.account_type}, " \return f"<AAAAAA(id={self.id}, uid={self.uid}, mid={self.mid}, account_type={self.account_type}, " \f"variety_code={self.variety_code}, balance={self.balance}, available={self.available}, " \f"freeze={self.freeze}, create_time={self.create_time}, update_time={self.update_time})>"# 3. 配置数据库连接信息
# 请根据实际情况修改数据库连接 URL 格式:
# 例如 'mysql+mysqlconnector://username:password@host:port/database_name'
DATABASE_URL = 'mysql+pymysql://user:123455hGv@172.31.116.214:3306/passwd'# 4. 创建数据库引擎
engine = create_engine(DATABASE_URL, echo=True)# 5. 创建所有表（如果尚未创建）
Base.metadata.create_all(engine)# 6. 创建 Session 类
Session = sessionmaker(bind=engine)# 7. 插入数据的操作
def insert_data(data):session = Session()# 创建 User 对象并添加到 sessionnew_data = AAAAAA(#id=1306936888,uid=data,mid=1681530298,account_type=1,variety_code='usdt',balance=200.00000000,available=200.00000000,freeze=0.00000000,create_time=datetime.strptime('2024-10-16 14:48:15', '%Y-%m-%d %H:%M:%S'),update_time=datetime.strptime('2024-11-05 11:09:30', '%Y-%m-%d %H:%M:%S'))session.add(new_data)# 提交事务session.commit()# 输出插入的对象print(f"Inserted: {new_data}")# 关闭 sessionsession.close()# def update_data(id):
#     session = Session()
#
#     # 7. 查询是否存在具有该 id 的记录
#     record = session.query(AAAAAA).filter(AAAAAA.id == id).first()#     if record:
#         # 8. 修改记录的字段
#         record.balance = 500.00000000  # 修改余额为 500
#         record.available = 500.00000000  # 修改可用余额为 500
#         record.freeze = 0.00000000  # 修改冻结余额为 0
#         record.update_time = datetime.strptime('2024-12-26 10:00:00', '%Y-%m-%d %H:%M:%S')  # 更新时间
#
#         # 9. 提交事务
#         session.commit()
#         print(f"Updated: {record}")
#     else:
#         print(f"Record with id {id} not found!")
#
#     # 10. 关闭 session
#     session.close()if __name__ == "__main__":uid = "1918509385"insert_data(uid)

查看全文

http://www.lryc.cn/news/510124.html