feat: Querying snippet list supports is_published for using snippets.py in workflow.

This commit is contained in:
FFXN 2026-03-05 17:13:20 +08:00
parent b160dce4db
commit b88195c7d9
3 changed files with 9 additions and 2 deletions

View File

@ -9,6 +9,7 @@ class SnippetListQuery(BaseModel):
page: int = Field(default=1, ge=1, le=99999)
limit: int = Field(default=20, ge=1, le=100)
keyword: str | None = None
is_published: bool | None = Field(default=None, description="Filter by published status")
class IconInfo(BaseModel):

View File

@ -1,7 +1,7 @@
import logging
from flask import request
from flask_restx import Resource, marshal, marshal_with
from flask_restx import Resource, marshal
from sqlalchemy.orm import Session
from werkzeug.exceptions import NotFound
@ -59,6 +59,7 @@ class CustomizedSnippetsApi(Resource):
page=query.page,
limit=query.limit,
keyword=query.keyword,
is_published=query.is_published,
)
return {

View File

@ -7,9 +7,9 @@ from typing import Any
from sqlalchemy import func, select
from sqlalchemy.orm import Session, sessionmaker
from core.workflow.variables.variables import VariableBase
from core.workflow.enums import NodeType
from core.workflow.nodes.node_mapping import LATEST_VERSION, NODE_TYPE_CLASSES_MAPPING
from core.workflow.variables.variables import VariableBase
from extensions.ext_database import db
from libs.infinite_scroll_pagination import InfiniteScrollPagination
from models import Account
@ -48,6 +48,7 @@ class SnippetService:
page: int = 1,
limit: int = 20,
keyword: str | None = None,
is_published: bool | None = None,
) -> tuple[Sequence[CustomizedSnippet], int, bool]:
"""
Get paginated list of snippets with optional search.
@ -56,6 +57,7 @@ class SnippetService:
:param page: Page number (1-indexed)
:param limit: Number of items per page
:param keyword: Optional search keyword for name/description
:param is_published: Optional filter by published status (True/False/None for all)
:return: Tuple of (snippets list, total count, has_more flag)
"""
stmt = (
@ -69,6 +71,9 @@ class SnippetService:
CustomizedSnippet.name.ilike(f"%{keyword}%") | CustomizedSnippet.description.ilike(f"%{keyword}%")
)
if is_published is not None:
stmt = stmt.where(CustomizedSnippet.is_published == is_published)
# Get total count
count_stmt = select(func.count()).select_from(stmt.subquery())
total = db.session.scalar(count_stmt) or 0