45 lines
1.5 KiB
Python
45 lines
1.5 KiB
Python
from fastapi import HTTPException
|
|
from sqlalchemy.orm import Session
|
|
|
|
from Model.AppModel import App
|
|
from Schemas import AppSchemas
|
|
|
|
|
|
def get_app_by_id(db: Session, app_id):
|
|
app = db.query(App).filter_by(id=app_id).first()
|
|
return app
|
|
|
|
|
|
def add_app(db: Session, app_id: str, name: str, secret_key: str, allowed: bool = False) -> App:
|
|
if db.query(App).filter_by(name=name).first():
|
|
raise HTTPException(status_code=403, detail="应用名称重复")
|
|
if db.query(App).filter_by(id=app_id).first():
|
|
raise HTTPException(status_code=403, detail="应用ID重复")
|
|
app = App(id=app_id, secret_key=secret_key, name=name, allowed=allowed)
|
|
db.add(app)
|
|
db.commit()
|
|
db.flush(app)
|
|
return app
|
|
|
|
|
|
def delete_app(db: Session, app_id):
|
|
db.query(App).filter_by(id=app_id).delete()
|
|
db.commit()
|
|
|
|
|
|
def change_app(db: Session, app_id: str, new_data: dict):
|
|
db.query(App).filter_by(ip=app_id).update(new_data)
|
|
db.commit()
|
|
|
|
|
|
def get_app_list(db: Session, query_params: AppSchemas.AppQueryParams):
|
|
query = db.query(App)
|
|
for key, value in query_params.dict().items():
|
|
if key not in ['page', 'page_size'] and value is not None:
|
|
if type(value) == str:
|
|
query = query.filter(getattr(App, key).like(f'%{value}%'))
|
|
count = query.count()
|
|
if query_params.page is not None and query_params.page_size is not None:
|
|
query = query.offset((query_params.page - 1) * query_params.page_size).limit(query_params.page_size).all()
|
|
return count,query
|