python使用sqlalchemy连接postgresql数据库

快来打我* 2022-10-27 13:55 752阅读 0赞

安装psycopg2和sqlalchemy模块：

pip install sqlalchemy
    pip install psycopg2

测试代码：

>     from sqlalchemy import Column, String, create_engine, Integer, Text
>     from sqlalchemy.orm import sessionmaker
>     from sqlalchemy.ext.declarative import declarative_base
>     import time
>     
>     # 创建对象的基类:
>     Base = declarative_base()
>     
>     
>     # 定义User对象:
>     class User(Base):
>         # 表的名字:
>         __tablename__ = 'users'
>     
>         # 表的结构:
>         id = Column(Integer, autoincrement=True, primary_key=True, unique=True, nullable=False)
>         name = Column(String(50), nullable=False)
>         sex = Column(String(4), nullable=False)
>         nation = Column(String(20), nullable=False)
>         birth = Column(String(8), nullable=False)
>         id_address = Column(Text, nullable=False)
>         id_number = Column(String(18), nullable=False)
>         creater = Column(String(32))
>         create_time = Column(String(20), nullable=False)
>         updater = Column(String(32))
>         update_time = Column(String(20), nullable=False, default=time.strftime("%Y-%m-%d %H:%M:%S", time.localtime()),
>                              onupdate=time.strftime("%Y-%m-%d %H:%M:%S", time.localtime()))
>         comment = Column(String(200))
>     
>     
>     # 初始化数据库连接:
>     engine = create_engine('postgres://postgres:123123@localhost:5432/postgres')  # 用户名:密码@localhost:端口/数据库名
>     
>     # 创建DBSession类型:
>     DBSession = sessionmaker(bind=engine)
>     
>     def createTable():
>         # 创建表
>         Base.metadata.create_all(engine)
>     
>     
>     def insertData():
>         # 插入操作
>         # 创建会话
>         session = DBSession()
>         # 创建新User对象:
>         local_time = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime())
>         new_user = User(name='hsh4', sex='男', nation='汉', birth='19991012', id_address='广州大学', id_number='441242142142',
>                         create_time=local_time)
>         # 添加到session:
>         session.add(new_user)
>         # 提交即保存到数据库:
>         session.commit()
>         # 关闭session:
>         session.close()
>     
>     def selectData():
>         # 查询操作
>         # 创建Session
>         session = DBSession()
>         # 创建Query查询，filter是where条件，最后调用one()返回唯一行，如果调用all()则返回所有行:
>         user = session.query(User).filter(User.id == '1' and User.name == 'hsh4').one()
>         print('name:', user.name)
>         print('id_address:', user.id_address)
>         session.close()  # 关闭Session
>     
>     def updateData():
>         # 更新操作
>         session = DBSession()  # 创建会话
>         users = session.query(User).filter_by(name="hsh4").first()  # 查询条件
>         users.id_number = "abcd"  # 更新操作
>         session.add(users)  # 添加到会话
>         session.commit()  # 提交即保存到数据库
>         session.close()  # 关闭会话
>     
>     def deleteData():
>         # 删除操作
>         session = DBSession()  # 创建会话
>         delete_users = session.query(User).filter(User.id == "1").first()
>         if delete_users:
>             session.delete(delete_users)
>             session.commit()
>         session.close()  # 关闭会话
>     
>     def dropTable():
>         # 删除表
>         session = DBSession()  # 创建会话
>         session.execute('drop table users')
>         session.commit()
>         session.close()
>     
>     
>     # createTable()
>     # insertData()
>     # selectData()
>     # updateData()
>     # deleteData()
>     # dropTable()

更多详细用法请参考官方文档：[PostgreSQL--SQLALchemy Documentation][]

以及ORM相关知识：[ORM知识介绍][ORM]

[PostgreSQL--SQLALchemy Documentation]: https://docs.sqlalchemy.org/en/13/dialects/postgresql.html
[ORM]: https://www.cnblogs.com/wj-1314/p/10627828.html