总结一下使用数据库的项目,使用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
网友评论