Class BookstackDataSource

app/data_source/sources/bookstack/bookstack.py:109–195

class BookstackDataSource(BaseDataSource):
    @staticmethod
    def get_config_fields() -> List[ConfigField]:
        return [
            ConfigField(label="BookStack instance URL", name="url"),
            ConfigField(label="Token ID", name="token_id", input_type=HTMLInputType.PASSWORD),
            ConfigField(label="Token Secret", name="token_secret", input_type=HTMLInputType.PASSWORD)
        ]

    @classmethod
    def get_display_name(cls) -> str:
        return "BookStack"

    @staticmethod
    def list_books(book_stack: BookStack) -> List[Dict]:
        # Usually the book_stack connection fails, so we retry a few times
        retries = 3
        for i in range(retries):
            try:
                return book_stack.get_all_books()
            except Exception as e:
                logging.error(f"BookStack connection failed: {e}")
                if i == retries - 1:
                    raise e

    @staticmethod
    async def validate_config(config: Dict) -> None:
        try:
            parsed_config = BookStackConfig(**config)
            book_stack = BookStack(url=parsed_config.url, token_id=parsed_config.token_id,
                                   token_secret=parsed_config.token_secret)
            BookstackDataSource.list_books(book_stack=book_stack)
        except Exception as e:
            raise InvalidDataSourceConfig from e

    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        book_stack_config = BookStackConfig(**self._raw_config)
        self._book_stack = BookStack(url=book_stack_config.url, token_id=book_stack_config.token_id,
                                     token_secret=book_stack_config.token_secret)

    def _list_books(self) -> List[Dict]:
        logger.info("Listing books with BookStack")
        return BookstackDataSource.list_books(book_stack=self._book_stack)

    def _feed_new_documents(self) -> None:
        logger.info("Feeding new documents with BookStack")

        books = self._list_books()
        for book in books:
            self.add_task_to_queue(self._feed_book, book=book)

    def _feed_book(self, book: Dict):
        logger.info(f"Getting documents from book {book['name']} ({book['id']})")
        pages = self._book_stack.get_all_pages_from_book(book)
        for page in pages:
            self.add_task_to_queue(self._feed_page, raw_page=page)