152 lines
4.4 KiB
Python
152 lines
4.4 KiB
Python
from sqlalchemy import func, select
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from app.models.user_oauth_app import UserOAuthApp
|
|
from app.utils.id_gen import generate_id
|
|
|
|
|
|
async def list_user_oauth_apps(
|
|
db: AsyncSession,
|
|
page: int = 1,
|
|
page_size: int = 20,
|
|
open_type: int | None = None,
|
|
status: int | None = None,
|
|
create_by: str | None = None,
|
|
app_id: str | None = None,
|
|
) -> dict:
|
|
query = select(UserOAuthApp).where(UserOAuthApp.deleted_at.is_(None)).order_by(UserOAuthApp.created_at.desc())
|
|
|
|
if open_type is not None:
|
|
query = query.where(UserOAuthApp.open_type == open_type)
|
|
|
|
if status is not None:
|
|
query = query.where(UserOAuthApp.status == status)
|
|
|
|
#if create_by is not None:
|
|
# query = query.where(UserOAuthApp.create_by == create_by)
|
|
|
|
if app_id is not None:
|
|
query = query.where(UserOAuthApp.app_id.like(f"%{app_id}%"))
|
|
|
|
count_query = select(func.count(UserOAuthApp.id)).where(UserOAuthApp.deleted_at.is_(None))
|
|
if open_type is not None:
|
|
count_query = count_query.where(UserOAuthApp.open_type == open_type)
|
|
if status is not None:
|
|
count_query = count_query.where(UserOAuthApp.status == status)
|
|
if create_by is not None:
|
|
count_query = count_query.where(UserOAuthApp.create_by == create_by)
|
|
if app_id is not None:
|
|
count_query = count_query.where(UserOAuthApp.app_id.like(f"%{app_id}%"))
|
|
total_result = await db.execute(count_query)
|
|
total = total_result.scalar() or 0
|
|
|
|
result = await db.execute(query.offset((page - 1) * page_size).limit(page_size))
|
|
items = result.scalars().all()
|
|
|
|
return {
|
|
"total": total,
|
|
"page": page,
|
|
"page_size": page_size,
|
|
"items": items,
|
|
}
|
|
|
|
|
|
async def get_user_oauth_app_by_id(db: AsyncSession, id: str) -> UserOAuthApp | None:
|
|
result = await db.execute(
|
|
select(UserOAuthApp).where(UserOAuthApp.id == id, UserOAuthApp.deleted_at.is_(None)).limit(1)
|
|
)
|
|
return result.scalar_one_or_none()
|
|
|
|
|
|
async def get_user_oauth_app_by_app_id(db: AsyncSession, app_id: str) -> UserOAuthApp | None:
|
|
result = await db.execute(
|
|
select(UserOAuthApp).where(UserOAuthApp.app_id == app_id, UserOAuthApp.deleted_at.is_(None)).limit(1)
|
|
)
|
|
return result.scalar_one_or_none()
|
|
|
|
|
|
async def create_user_oauth_app(
|
|
db: AsyncSession,
|
|
app_id: str,
|
|
secret: str,
|
|
open_type: int,
|
|
create_by: str | None = None,
|
|
count: int = 100,
|
|
auth_url: str | None = None,
|
|
company: str | None = None,
|
|
) -> UserOAuthApp:
|
|
existing = await get_user_oauth_app_by_app_id(db, app_id)
|
|
if existing:
|
|
raise ValueError("应用id已存在")
|
|
|
|
app = UserOAuthApp(
|
|
id=generate_id(),
|
|
app_id=app_id,
|
|
secret=secret,
|
|
open_type=open_type,
|
|
max_count=count,
|
|
auth_url=auth_url,
|
|
company=company,
|
|
create_by=create_by,
|
|
)
|
|
db.add(app)
|
|
await db.flush()
|
|
await db.commit()
|
|
await db.refresh(app)
|
|
return app
|
|
|
|
|
|
async def update_user_oauth_app(
|
|
db: AsyncSession,
|
|
id: str,
|
|
secret: str | None = None,
|
|
open_type: int | None = None,
|
|
status: int | None = None,
|
|
count: int | None = None,
|
|
auth_url: str | None = None,
|
|
company: str | None = None,
|
|
create_by: str | None = None,
|
|
) -> UserOAuthApp | None:
|
|
app = await get_user_oauth_app_by_id(db, id)
|
|
if not app:
|
|
return None
|
|
|
|
if secret is not None:
|
|
app.secret = secret
|
|
if open_type is not None:
|
|
app.open_type = open_type
|
|
if status is not None:
|
|
app.status = status
|
|
if count is not None:
|
|
app.max_count = count
|
|
if auth_url is not None:
|
|
app.auth_url = auth_url
|
|
if company is not None:
|
|
app.company = company
|
|
if create_by is not None:
|
|
app.create_by = create_by
|
|
|
|
await db.flush()
|
|
await db.commit()
|
|
await db.refresh(app)
|
|
return app
|
|
|
|
|
|
async def delete_user_oauth_app(db: AsyncSession, id: str, create_by: str | None = None) -> bool:
|
|
app = await get_user_oauth_app_by_id(db, id)
|
|
if not app:
|
|
return False
|
|
|
|
app.deleted_at = func.now()
|
|
if create_by is not None:
|
|
app.create_by = create_by
|
|
await db.flush()
|
|
await db.commit()
|
|
return True
|
|
|
|
|
|
async def get_apps_by_open_type(db: AsyncSession, open_type: int) -> list[UserOAuthApp]:
|
|
result = await db.execute(
|
|
select(UserOAuthApp).where(UserOAuthApp.open_type == open_type, UserOAuthApp.deleted_at.is_(None))
|
|
)
|
|
return result.scalars().all() |