Python sqlalchemy使用
2023-12-14 11:51:35
基本结构
#!/usr/bin/python3
# -*- coding:utf-8 -*-
"""
@author: JHC
@file: base_db.py
@time: 2023/6/19 21:34
@desc:
"""
from sqlalchemy import create_engine,text
from sqlalchemy.orm import sessionmaker,scoped_session
from contextlib import contextmanager
from sqlalchemy.pool import QueuePool
from sdk.base.base_tables import Base
class DB(object):
"""
"""
def __init__(self,username,password,host,port,db,poll_size=50,debug=True):
self.username = username
self.password = password
self.host = host
self.port = port
self.db = db
self.poll_size = poll_size
# 初始化数据库连接池:
self.engine = create_engine(f'mysql+mysqlconnector'
f'://{self.username}:{self.password}'
f'@{self.host}:{self.port}/{self.db}',
poolclass=QueuePool,pool_size=self.poll_size,echo=True if debug else False)
self.__create_db()
# 创建线程安全的DBSession类型:
self.db_session = scoped_session(sessionmaker(bind=self.engine, autocommit=False))
def __create_db(self):
"""
新建表
:return:
"""
Base.metadata.create_all(self.engine, checkfirst=True)
def run_origin_sql(self,session,sql):
"""
执行原生sql
:param session:
:param sql:
:return:
"""
return session.execute(text(f"""{sql}"""))
@contextmanager
def get_session(self):
"""
返回数据库连接对象
:return:
"""
session = self.db_session()
try:
yield session
session.commit()
except:
session.rollback()
raise
finally:
session.close()
base_tables.py
# !/usr/bin/python3
# -*- coding:utf-8 -*-
"""
@author: JHC000abc@gmail.com
@file: base_tables.py
@time: 2023/12/9 21:57
@desc:
"""
from sqlalchemy.ext.declarative import declarative_base
# 创建对象的基类,单写出来,避免循环引用
Base = declarative_base()
使用样例
from sdk.base.base_db import *
from db_tables.KDQ_USER_INFO import *
from db_tables.KDQ_CONTENT_INFO import *
from db_tables.KDQ_COMMENT_INFO import *
from test1 import *
import time
db = DB(host="127.0.0.1", port=3306, username="root", password="123456", db=DB_KDQ)
with db.get_session() as session:
session.***
from db_tables.KDQ_USER_INFO import *
from db_tables.KDQ_CONTENT_INFO import *
from db_tables.KDQ_COMMENT_INFO import *
这三个是写的数据表表类
差不多长这样
# !/usr/bin/python3
# -*- coding:utf-8 -*-
"""
@author: JHC000abc@gmail.com
@file: KDQ_USER_INFO.py
@time: 2023/12/9 21:54
@desc:
"""
from sqlalchemy.sql import func
from sqlalchemy import Column,INT,VARCHAR,TEXT,Boolean,DateTime
from sdk.base.base_tables import Base
class UserInfo(Base):
"""
用户信息表
"""
__tablename__ = 'user_info'
id = Column(INT,primary_key=True,autoincrement=True,nullable=False,index=True,unique=True)
user_id = Column(INT,nullable=False,unique=True)
user_name = Column(VARCHAR(50),nullable=False)
nick_name = Column(VARCHAR(255),nullable=False)
real_user_info = Column(VARCHAR(255),nullable=False)
avatar = Column(TEXT,nullable=True)
role_set = Column(TEXT,nullable=True)
permission_set = Column(TEXT,nullable=True)
email = Column(VARCHAR(100),nullable=False,default=0)
mobile = Column(VARCHAR(11),nullable=False,default=0)
gender = Column(VARCHAR(10),nullable=False)
use_default_password = Column(Boolean,nullable=True,default=False)
comment_num = Column(INT,nullable=False,default=0)
like_num = Column(INT,nullable=False,default=0)
view_num = Column(INT,nullable=False,default=0)
follower_num = Column(INT,nullable=False,default=0)
mute_num = Column(INT,nullable=False,default=0)
create_time = Column(DateTime,nullable=False,default=func.now())
update_time = Column(DateTime,nullable=False,default=func.now(),onupdate=func.now())
__table_args__ = {
'mysql_charset': 'utf8mb4' # 设置字符集为utf8mb4
}
DB_KDQ = "demo"
文章来源:https://blog.csdn.net/CXY00000/article/details/134914809
本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。 如若内容造成侵权/违法违规/事实不符,请联系我的编程经验分享网邮箱:veading@qq.com进行投诉反馈,一经查实,立即删除!
本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。 如若内容造成侵权/违法违规/事实不符,请联系我的编程经验分享网邮箱:veading@qq.com进行投诉反馈,一经查实,立即删除!