Skip to content
Closed
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
6 changes: 6 additions & 0 deletions docs/api.rst
Original file line number Diff line number Diff line change
Expand Up @@ -103,6 +103,12 @@ Search Handlers
.. autoclass:: flogin.search_handler.SearchHandler
:members:

.. attributetable:: flogin.search_group.SearchGroup

.. autoclass:: flogin.search_group.SearchGroup
:members:
:exclude-members: callback, condition

.. _builtin_search_conditions:

Builtin Search Conditions
Expand Down
1 change: 1 addition & 0 deletions flogin/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -8,6 +8,7 @@
from .jsonrpc import *
from .plugin import *
from .query import *
from .search_group import *
from .search_handler import *
from .settings import *
from .utils import VersionInfo
Expand Down
228 changes: 228 additions & 0 deletions flogin/search_group.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,228 @@
from __future__ import annotations

from typing import TYPE_CHECKING, Any, Generic, Self

from ._types import (
PluginT,
SearchHandlerCallback,
SearchHandlerCallbackReturns,
)
Comment on lines +5 to +9
Copy link

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

💡 Verification agent

⚠️ Potential issue

🧩 Analysis chain

Fix import error for PluginT.

The pipeline is failing with ImportError: cannot import name 'PluginT' from 'flogin._types'. This needs to be fixed for the code to work.


🏁 Script executed:

#!/bin/bash
# Check if PluginT is defined in _types.py
rg -A 5 'PluginT' | grep -v 'search_group.py'

Length of output: 17092


I'll take a closer look at the contents of flogin/_types.py to understand the conditional definition of PluginT and why it might not be available when imported by search_group.py. Let me inspect the file.


🏁 Script executed:

#!/bin/bash
# Output the first 100 lines of flogin/_types.py to inspect definitions for PluginT.
sed -n '1,100p' flogin/_types.py

Length of output: 773


Investigated conditional definition of PluginT in flogin/_types.py.

The file clearly defines PluginT in both branches of the TYPE_CHECKING condition—when type checking (using the advanced definition with bounds and covariance) and at runtime (a plain TypeVar). This indicates that PluginT is indeed available from flogin/_types.py. The pipeline’s ImportError appears to be caused by a circular dependency issue during type checking—specifically, when TYPE_CHECKING is True, _types.py attempts to import Plugin from flogin/plugin.py, and a circular import may be preventing PluginT from being exported properly.

To resolve this, please consider one of the following fixes:

  • Move the definition of PluginT outside of the TYPE_CHECKING block so that it’s unconditionally available. For example, always define PluginT with the advanced TypeVar using importlib “if TYPE_CHECKING:” only for imports needed solely for type annotations. This avoids a circular dependency when the type checker is active.

  • Alternatively, if you need to keep the conditional definition, restructure your code (and possibly the import order in flogin/plugin.py) to break the circular dependency.

Please verify which approach best fits the overall design of our codebase.

🧰 Tools
🪛 GitHub Actions: build

[error] 5-5: ImportError: cannot import name 'PluginT' from 'flogin._types' (unknown location). Did you mean: 'plugin'?

from .jsonrpc.results import Result
from .search_handler import SearchHandler
from .utils import (
MISSING,
decorator,
)

if TYPE_CHECKING:
from .query import Query

__all__ = ("SearchGroup",)


class SearchGroup(SearchHandler[PluginT], Generic[PluginT]):
r"""A subclass of :class:`~flogin.search_handler.SearchHandler` to let you easily create a nested command structure.

The keywords in the constructor can also be passed into the subclassed init, like so: ::

class MyGroup(SearchGroup, prefix="text"):
...

# is equal to

class MyGroup(SearchGroup):
def __init__(self):
super().__init__(prefix="text")

.. versionadded:: 2.0.0

Example
-------
Adding a subcommand named ``cmd``

.. code-block:: py3
:linenos:

from flogin import SearchGroup

group = SearchGroup("foo")

@group.sub("cmd")
async def cmd(query):
return "Hi from cmd"

# cmd can be invoked with "foo cmd"

Adding nested subgroups

.. code-block:: py3
:linenos:

from flogin import SearchGroup

group = SearchGroup("foo")
sub = group.sub("sub")

@sub.sub("cmd2")
async def cmd2(query):
return "Hi from cmd2 under cmd1"

# cmd2 can be invoked with "foo sub cmd2"

Attributes
----------
prefix: :class:`str`
The prefix used to trigger the search group
sep: :class:`str`
The character that seperates the group's prefix and subhandler's prefixes during validation. Defaults to a space.
"""

prefix: str
sep: str = " "

def __init__(self, prefix: str = MISSING, sep: str = MISSING) -> None:
if prefix is not MISSING:
self.prefix = prefix
else:
if not hasattr(self, "prefix"):
raise ValueError(
"Prefix arg must be provided in __init__ or __init_subclass__"
)
if sep is not MISSING:
self.sep = sep

self.parent: SearchGroup | None = None
self._subgroups: dict[str, SearchGroup] = {}

def __init_subclass__(cls, prefix: str = MISSING, sep: str = MISSING) -> None:
cls.prefix = prefix
cls.sep = " " if sep is MISSING else sep

def condition(self, query: Query) -> bool:
try:
return f"{query.text}{self.sep}".startswith(f"{self.prefix}{self.sep}")
except IndexError:
return False

@property
def subgroups(self) -> dict[str, SearchGroup]:
r"""dict[:class:`str`, :class:`~flogin.search_group.SearchGroup`]: A copied version of the subgroups that have been registered to this group."""
return self._subgroups.copy()

@property
def signature(self) -> str:
r""":class:`str` the group's signature. This property will grab all of the group's parent's prefixes, and join them together using the :attr:`~flogin.search_group.SearchGroup.sep` attribute"""

parts = []
parent = self

while parent:
parts.append(parent.prefix)
parent = parent.parent
return self.sep.join(reversed(parts))

def create_result(self, key: str, query: Query) -> Result:
r"""This is a method that is used by the default root callback and can be overriden.

It is used to turn the key of one of the subhandlers into a result object that is returned. The default result that is returned has the title set to the key, with the ``auto_complete_text`` and callback set to properly change the query to trigger the subhander.

Parameters
----------
key: :class:`str`
The key to generate the result for
query: :class:`~flogin.query.Query`
The query that this is being generated for

Returns
-------
:class:`~flogin.jsonrpc.results.Result`
The generated result object
"""

assert self.plugin
assert self.plugin.last_query

new = f"{self.plugin.last_query.keyword} {self.signature}{self.sep}{key}{self.sep}"

async def callback() -> bool:
assert self.plugin

await self.plugin.api.change_query(new)
return False

return Result.create_with_partial(
callback,
title=key,
auto_complete_text=new,
)

async def root_handler(self, query: Query) -> list[Result]:
return [self.create_result(key, query) for key in self._subgroups]

def callback(self, query: Query) -> SearchHandlerCallbackReturns:
parts = query.text.split(self.sep)
query._data["search"] = query.text.removeprefix(self.prefix).removeprefix(
self.sep
)

try:
handler = self._subgroups[parts[1]]
except (KeyError, IndexError):
handler = self.root_handler

if isinstance(handler, SearchGroup):
handler.plugin = self.plugin
handler = handler.callback

return handler(query)
Comment on lines +163 to +177
Copy link

@coderabbitai coderabbitai bot Feb 6, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🛠️ Refactor suggestion

Add error handling for invalid query format.

The callback method splits the query text but doesn't validate the format. This could lead to unexpected behavior with malformed queries.

Consider adding validation:

def callback(self, query: Query) -> SearchHandlerCallbackReturns:
+   if not query.text:
+       return self.root_handler(query)
+
    parts = query.text.split(self.sep)
    query._data["search"] = query.text.removeprefix(self.prefix).removeprefix(
        self.sep
    )

    try:
        handler = self._subgroups[parts[1]]
    except (KeyError, IndexError):
        handler = self.root_handler

    if isinstance(handler, SearchGroup):
        handler.plugin = self.plugin
        handler = handler.callback

    return handler(query)
📝 Committable suggestion

‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.

Suggested change
parts = query.text.split(self.sep)
query._data["search"] = query.text.removeprefix(self.prefix).removeprefix(
self.sep
)
try:
handler = self._subgroups[parts[1]]
except (KeyError, IndexError):
handler = self.root_handler
if isinstance(handler, SearchGroup):
handler.plugin = self.plugin
handler = handler.callback
return handler(query)
def callback(self, query: Query) -> SearchHandlerCallbackReturns:
if not query.text:
return self.root_handler(query)
parts = query.text.split(self.sep)
query._data["search"] = query.text.removeprefix(self.prefix).removeprefix(self.sep)
try:
handler = self._subgroups[parts[1]]
except (KeyError, IndexError):
handler = self.root_handler
if isinstance(handler, SearchGroup):
handler.plugin = self.plugin
handler = handler.callback
return handler(query)

Copy link
Owner Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

can you explain this more? I don't understand why this change would help

Copy link

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

@cibere Let me explain why this change would be helpful:

Currently, when an empty query is received, the code still tries to:

  1. Split the empty string ("") into parts
  2. Remove prefixes from an empty string
  3. Try to access parts[1] which fails because parts is just [""]
  4. Finally falls back to root_handler due to the IndexError

The suggested change simply checks for empty query upfront and returns root_handler immediately. While both approaches end up calling root_handler, the suggested way:

  • Is more explicit about handling empty queries
  • Avoids unnecessary string operations
  • Makes the code's intent clearer

Would you like me to show some example scenarios to demonstrate this?

Copy link
Owner Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

yes please

Copy link

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

@cibere Let me demonstrate with two scenarios:

Scenario 1: Current Behavior (Empty Query)

query.text = ""

# Current flow:
parts = "".split(" ")  # parts = [""]
query._data["search"] = "".removeprefix(self.prefix).removeprefix(self.sep)  # Unnecessary operations
try:
    handler = self._subgroups[parts[1]]  # IndexError because parts only has one item
except (KeyError, IndexError):
    handler = self.root_handler  # Finally reaches root_handler

Scenario 2: Proposed Behavior (Empty Query)

query.text = ""

# Proposed flow:
if not query.text:
    return self.root_handler(query)  # Immediately returns root_handler

The end result is the same (calling root_handler), but the proposed way:

  1. Skips unnecessary string operations
  2. Avoids throwing and catching exceptions
  3. Makes it immediately clear that empty queries show the root menu

Would this help explain the reasoning better?


def __call__(self, callback: SearchHandlerCallback) -> Self:
setattr(self, "root_handler", callback)
return self

@decorator(is_factory=True)
def sub(self, prefix: str) -> SearchGroup:
r"""Adds a subgroup to the search group.

Parameters
----------
prefix: :class:`str`
The prefix used to trigger the subgroup

Raises
------
:class:`ValueError`
This is raised when the given prefix is already associated with a subgroup.

Example
-------
.. code-block:: py3
:linenos:

group = SearchGroup(...)

@group.sub("prefix")
async def subgroup(query):
...
"""

if prefix in self._subgroups:
raise ValueError(
f"A subgroup with the {prefix!r} prefix has already been registered"
)

group = SearchGroup(prefix, self.sep)
group.parent = self
self._subgroups[prefix] = group
return group

def get_tree(self) -> dict[str, Any]:
return {
"prefix": self.prefix,
"children": {
name: value.get_tree()
if isinstance(value, SearchGroup)
else repr(value)
for name, value in self._subgroups.items()
},
}
9 changes: 7 additions & 2 deletions flogin/utils.py
Original file line number Diff line number Diff line change
Expand Up @@ -298,9 +298,14 @@ def inner(func: T) -> T:


class func_with_self(Generic[P, ReturnT, OwnerT]):
def __init__(self, func: Callable[Concatenate[OwnerT, P], ReturnT]) -> None:
def __init__(
self,
func: Callable[Concatenate[OwnerT, P], ReturnT],
*,
owner: OwnerT = MISSING,
) -> None:
self.func = func
self.owner: OwnerT | None = None
self.owner: OwnerT | None = owner if owner is not MISSING else None

update_wrapper(wrapped=func, wrapper=self)

Expand Down
Loading