search_service.py

Overview

The search_service.py file defines the SearchService class, which provides data access and business logic related to "Search" entities within the InfiniFlow application. It acts as a service layer interfacing with the database models to perform CRUD operations, filtering, pagination, access control, and detailed retrieval of search records. This service builds upon a common base service and integrates tightly with user and status models to ensure data integrity and permission checks.

Key functionalities include:

• Creating and saving new search records with automatic timestamp management.

• Checking if a search record is accessible for deletion by a given user.

• Retrieving detailed information of a single search entry, including tenant user info.

• Querying search records filtered by tenant IDs, user ownership, keywords, and supporting pagination and sorting.

This service plays a crucial role in managing "Search" objects in the system, encapsulating database interactions and enforcing business rules.

Class: SearchService

Inherits from: CommonService

SearchService is a specialized service class dedicated to managing Search model objects. It extends CommonService to reuse common database interaction patterns and adds methods specific to the search domain.

Properties

• model: Class attribute set to the Search database model. This binds the service to operate on Search records.

Methods

save(**kwargs)

Creates and saves a new Search record in the database.

• Parameters:

  • **kwargs (dict): Arbitrary keyword arguments representing fields of the Search model to be created.

• Returns:

  • Search: The newly created Search model instance.

• Details:

  • Automatically sets the following fields:

    • create_time: Current timestamp (int or float).

    • create_date: Current date formatted as string.

    • update_time: Current timestamp.

    • update_date: Current date formatted as string.

  • Calls the model's create method with the augmented fields.

• Usage Example:

  new_search = SearchService.save(
      name="New Search",
      description="A description",
      created_by=123,
      tenant_id=456,
      status=StatusEnum.VALID.value,
      search_config={"query": "some config"},
      avatar="avatar_url"
  )
  

accessible4deletion(search_id: int, user_id: int) -> bool

Checks if a specific Search record is accessible for deletion by a particular user.

• Parameters:

  • search_id (int): The unique ID of the Search record.

  • user_id (int): The ID of the user requesting deletion.

• Returns:

  • bool: True if the user created the search and it has a valid status, otherwise False.

• Implementation Details:

  • Queries the Search table filtering by:

    • id == search_id

    • created_by == user_id

    • status == StatusEnum.VALID

  • Uses Peewee's .first() to check for existence.

• Usage Example:

  can_delete = SearchService.accessible4deletion(search_id=10, user_id=42)
  if can_delete:
      # Proceed with deletion
  

get_detail(search_id: int) -> dict

Retrieves detailed information for a single Search record, including related tenant user data.

• Parameters:

  • search_id (int): The unique identifier of the Search record.

• Returns:

  • dict: A dictionary containing search fields and related tenant user information. Returns empty dict if not found.

• Fields Returned:

  • id, avatar, tenant_id, name, description, created_by, search_config, update_time

  • Tenant user's nickname and tenant_avatar (aliased from User.avatar)

• Implementation Details:

  • Joins the User table on the tenant ID with a valid user status.

  • Filters by search ID and a valid status.

  • Returns the first matching record as a dictionary.

• Usage Example:

  search_info = SearchService.get_detail(search_id=15)
  if search_info:
      print(search_info["name"], search_info["nickname"])
  else:
      print("Search not found or invalid.")
  

get_by_tenant_ids(joined_tenant_ids: list[int], user_id: int, page_number: int, items_per_page: int, orderby: str, desc: bool, keywords: str) -> tuple[list[dict], int]

Queries and paginates search records filtered by tenants, user ownership, keywords, and sorted by specified fields.

• Parameters:

  • joined_tenant_ids (list[int]): List of tenant IDs to filter searches by.

  • user_id (int): The current user's ID, to include their own records.

  • page_number (int): The page number for pagination.

  • items_per_page (int): Number of items per page.

  • orderby (str): The field name to order the results by.

  • desc (bool): Whether to order descending (True) or ascending (False).

  • keywords (str): Optional search keyword to filter by search name (case insensitive).

• Returns:

  • tuple:

    • list[dict]: List of search records as dictionaries.

    • int: Total count of matched records before pagination.

• Returned Fields per Record:

  • id, avatar, tenant_id, name, description, created_by, status

  • update_time, create_time

  • Tenant user's nickname and tenant_avatar

• Implementation Details:

  • Filters searches where tenant is in joined_tenant_ids or tenant is the current user.

  • Filters for valid status only.

  • Applies keyword filtering on name using case-insensitive containment.

  • Applies ordering based on orderby and desc flags.

  • Uses Peewee's paginate method for pagination.

  • Returns both the list of results and the total count.

• Usage Example:

  searches, total = SearchService.get_by_tenant_ids(
      joined_tenant_ids=[1, 2, 3],
      user_id=5,
      page_number=1,
      items_per_page=10,
      orderby="update_time",
      desc=True,
      keywords="test"
  )
  print(f"Total searches found: {total}")
  for search in searches:
      print(search["name"], search["nickname"])
  

Important Implementation Details

• Uses Peewee ORM for all database interactions, leveraging query builders, joins, filters, and pagination.

• Timestamp fields (create_time, update_time) are managed via utility functions current_timestamp() and datetime_format().

• The service enforces status checks using StatusEnum.VALID to ensure only active/valid records are considered.

• Tenant user information is retrieved via a join on the User table filtered by valid user status.

• The get_by_tenant_ids method supports keyword search by applying a case-insensitive contains filter on Search.name.

• Sorting is flexible via dynamic getter method getter_by on the model, allowing ordering by arbitrary fields.

• The decorator @DB.connection_context() is used on methods that interact with the database, ensuring proper connection management.

Interaction with Other Parts of the System

• Models:

  • Search: Core database model representing search entities.

  • User: Used to fetch tenant/owner user details.

  • StatusEnum: Enum class providing status values, e.g., VALID.

• CommonService: Base service class providing shared methods and properties.

• Database Layer (api.db): Provides connection context and models.

• Utilities (api.utils): Provides timestamp and datetime formatting functions.

• Application Layer: This service is likely consumed by API endpoints or other business logic components to manage user searches.

Visual Diagram: Class Structure of SearchService

classDiagram
    class CommonService {
        <<abstract>>
        +model
        +save()
        +getter_by(field)
    }

    class SearchService {
        +model = Search
        +save(**kwargs)
        +accessible4deletion(search_id, user_id) bool
        +get_detail(search_id) dict
        +get_by_tenant_ids(joined_tenant_ids, user_id, page_number, items_per_page, orderby, desc, keywords) tuple[list[dict], int]
    }

    SearchService --|> CommonService

Summary

The search_service.py module encapsulates all business logic and database access patterns related to managing Search entities. It provides a clear interface for creating, querying, retrieving details, and validating permissions on search records. Leveraging Peewee ORM, it integrates robust querying capabilities with pagination, filtering, and sorting. This service is a foundational component for any feature that requires managing or displaying search configurations within the InfiniFlow platform.