python sqlalchemy项目结构

总结一下使用数据库的项目，使用sqlalchemy的一般结构

首先是连接数据库，单独使用一个文件进行连接

db\session.py

from sqlalchemy import create_engine, MetaData
from clickhouse_sqlalchemy import make_session, get_declarative_base

from tools.config_add import CONFIG

url = CONFIG.chsql.url
port = CONFIG.chsql.port
database = CONFIG.chsql.database
username = CONFIG.chsql.username
password = CONFIG.chsql.password
uri = "clickhouse://" + username + ":" + password + "@" + url + "/" + database

engine = create_engine(uri)
session = make_session(engine)
metadata = MetaData(bind=engine)

Base = get_declarative_base(metadata=metadata)

每个表定义一个类

models\chdb\dms_gateway_log.py

from sqlalchemy import Column
from clickhouse_sqlalchemy import types, engines

from db.session import Base, engine

class LogEntity(Base):
 __tablename__ = "dms_gateway_log"
 id = Column(types.Int, primary_key=True)
 login_id = Column(types.String)
 receive_time = Column(types.String)
 offline_content = Column(types.String)
 online_content = Column(types.String)
 online_file = Column(types.String)

 __table_args__ = (
 engines.Memory(),
    )

def chdb_init():   
 Base.metadata.create_all(engine)

定义crud服务

db\db_services\chdb\dms_log_service.py

from db.session import session
from models.chdb.dms_gateway_log import LogEntity

class Log_services():

 def add(self,id):
 logrecord = LogEntity(id = id, login_id='124411')
 session.add(logrecord)
 session.commit()

 def get(self, id):
 pp = session.query(LogEntity).filter(LogEntity.id==id).first()
 return pp.login_id