Initial commit

Author: Garulf

.flake8

@@ -0,0 +1,2 @@
+[flake8]
+max-line-length = 127

.gitignore

@@ -0,0 +1,164 @@
+# Byte-compiled / optimized / DLL files
+__pycache__/
+*.py[cod]
+*$py.class
+
+# C extensions
+*.so
+
+# Distribution / packaging
+.Python
+build/
+develop-eggs/
+dist/
+downloads/
+eggs/
+.eggs/
+lib/
+lib64/
+parts/
+sdist/
+var/
+wheels/
+share/python-wheels/
+*.egg-info/
+.installed.cfg
+*.egg
+MANIFEST
+
+# PyInstaller
+#  Usually these files are written by a python script from a template
+#  before PyInstaller builds the exe, so as to inject date/other infos into it.
+*.manifest
+*.spec
+
+# Installer logs
+pip-log.txt
+pip-delete-this-directory.txt
+
+# Unit test / coverage reports
+htmlcov/
+.tox/
+.nox/
+.coverage
+.coverage.*
+.cache
+nosetests.xml
+coverage.xml
+*.cover
+*.py,cover
+.hypothesis/
+.pytest_cache/
+cover/
+
+# Translations
+*.mo
+*.pot
+
+# Django stuff:
+*.log
+local_settings.py
+db.sqlite3
+db.sqlite3-journal
+
+# Flask stuff:
+instance/
+.webassets-cache
+
+# Scrapy stuff:
+.scrapy
+
+# Sphinx documentation
+docs/_build/
+
+# PyBuilder
+.pybuilder/
+target/
+
+# Jupyter Notebook
+.ipynb_checkpoints
+
+# IPython
+profile_default/
+ipython_config.py
+
+# pyenv
+#   For a library or package, you might want to ignore these files since the code is
+#   intended to run in multiple environments; otherwise, check them in:
+# .python-version
+
+# pipenv
+#   According to pypa/pipenv#598, it is recommended to include Pipfile.lock in version control.
+#   However, in case of collaboration, if having platform-specific dependencies or dependencies
+#   having no cross-platform support, pipenv may install dependencies that don't work, or not
+#   install all needed dependencies.
+#Pipfile.lock
+
+# poetry
+#   Similar to Pipfile.lock, it is generally recommended to include poetry.lock in version control.
+#   This is especially recommended for binary packages to ensure reproducibility, and is more
+#   commonly ignored for libraries.
+#   https://python-poetry.org/docs/basic-usage/#commit-your-poetrylock-file-to-version-control
+#poetry.lock
+
+# pdm
+#   Similar to Pipfile.lock, it is generally recommended to include pdm.lock in version control.
+#pdm.lock
+#   pdm stores project-wide configurations in .pdm.toml, but it is recommended to not include it
+#   in version control.
+#   https://pdm.fming.dev/#use-with-ide
+.pdm.toml
+
+# PEP 582; used by e.g. github.com/David-OConnor/pyflow and github.com/pdm-project/pdm
+__pypackages__/
+
+# Celery stuff
+celerybeat-schedule
+celerybeat.pid
+
+# SageMath parsed files
+*.sage.py
+
+# Environments
+.env
+.venv
+env/
+venv/
+ENV/
+env.bak/
+venv.bak/
+
+# Spyder project settings
+.spyderproject
+.spyproject
+
+# Rope project settings
+.ropeproject
+
+# mkdocs documentation
+/site
+
+# mypy
+.mypy_cache/
+.dmypy.json
+dmypy.json
+
+# Pyre type checker
+.pyre/
+
+# pytype static type analyzer
+.pytype/
+
+# Cython debug symbols
+cython_debug/
+
+# PyCharm
+#  JetBrains specific template is maintained in a separate JetBrains.gitignore that can
+#  be found at https://github.com/github/gitignore/blob/main/Global/JetBrains.gitignore
+#  and can be added to the global gitignore or merged into this file.  For a more nuclear
+#  option (not recommended) you can uncomment the following to ignore the entire idea folder.
+#.idea/
+
+.vscode
+
+test.py

README.md

@@ -0,0 +1,26 @@
+import os
+import logging
+
+from .plugin import Plugin, ResultResponse, send_results
+from .result import Result
+from .jsonrpc import JsonRPCRequest
+
+
+log_level = os.environ.get("FLOW_LAUNCHER_API_LOG_LEVEL", "INFO")
+
+logger = logging.getLogger(__name__)
+
+__all__ = [
+    "Plugin",
+    "ResultResponse",
+    "send_results",
+    "Result",
+    "JsonRPCRequest",
+]
+
+
+logging.basicConfig(
+    level=log_level,
+    format="%(asctime)s %(name)s %(levelname)s %(message)s",
+    datefmt="%Y-%m-%d %H:%M:%S",
+)

pyflowlauncher/__init__.py

@@ -0,0 +1,53 @@
+from .jsonrpc import JsonRPCRequest
+
+NAME_SPACE = 'Flow.Launcher'
+
+
+def change_query(query: str, requery: bool = False) -> JsonRPCRequest:
+    """Change the query in Flow Launcher."""
+    return {"method": f"{NAME_SPACE}.ChangeQuery", "parameters": [query, requery]}
+
+
+def shell_run(command: str) -> JsonRPCRequest:
+    """Run a shell command."""
+    return {"method": f"{NAME_SPACE}.ShellRun", "parameters": [command]}
+
+
+def close_app() -> JsonRPCRequest:
+    """Close Flow Launcher."""
+    return {"method": f"{NAME_SPACE}.CloseApp", "parameters": []}
+
+
+def hide_app() -> JsonRPCRequest:
+    """Hide Flow Launcher."""
+    return {"method": f"{NAME_SPACE}.HideApp", "parameters": []}
+
+
+def show_app() -> JsonRPCRequest:
+    """Show Flow Launcher."""
+    return {"method": f"{NAME_SPACE}.ShowApp", "parameters": []}
+
+
+def show_msg(title: str, sub_title: str, ico_path: str = "") -> JsonRPCRequest:
+    """Show a message in Flow Launcher."""
+    return {"method": f"{NAME_SPACE}.ShowMsg", "parameters": [title, sub_title, ico_path]}
+
+
+def open_setting_dialog() -> JsonRPCRequest:
+    """Open the settings window in Flow Launcher."""
+    return {"method": f"{NAME_SPACE}.OpenSettingDialog", "parameters": []}
+
+
+def start_loading_bar() -> JsonRPCRequest:
+    """Start the loading bar in Flow Launcher."""
+    return {"method": f"{NAME_SPACE}.StartLoadingBar", "parameters": []}
+
+
+def stop_loading_bar() -> JsonRPCRequest:
+    """Stop the loading bar in Flow Launcher."""
+    return {"method": f"{NAME_SPACE}.StopLoadingBar", "parameters": []}
+
+
+def reload_plugins() -> JsonRPCRequest:
+    """Reload the plugins in Flow Launcher."""
+    return {"method": f"{NAME_SPACE}.ReloadPlugins", "parameters": []}

pyflowlauncher/api.py

@@ -0,0 +1,16 @@
+
+
+class EventHandler:
+
+    def __init__(self):
+        self._methods = {}
+
+    def add_method(self, method, *, name=None):
+        self._methods[name or method.__name__] = method
+
+    def add_methods(self, methods):
+        for method in methods:
+            self.add_method(method)
+
+    def __call__(self, method, *args, **kwargs):
+        return self._methods[method](*args, **kwargs)

pyflowlauncher/event.py

@@ -0,0 +1,24 @@
+import json
+import sys
+from typing import Any, Dict, Mapping, TypedDict, NotRequired
+
+
+class JsonRPCRequest(TypedDict):
+    method: str
+    parameters: list
+    settings: NotRequired[Dict[Any, Any]]
+
+
+Response = Dict[Any, Any]
+
+
+class JsonRPCClient:
+
+    def send(self, data: Mapping) -> None:
+        json.dump(data, sys.stdout)
+
+    def recieve(self) -> JsonRPCRequest:
+        try:
+            return json.loads(sys.argv[1])
+        except IndexError:
+            return {'method': 'query', 'parameters': ['']}

pyflowlauncher/icons.py

@@ -0,0 +1,55 @@
+from typing import Any, Dict, Iterable, Callable, Optional, TypedDict, Union
+from functools import wraps
+
+from .result import Result
+from .jsonrpc import JsonRPCClient
+from .event import EventHandler
+from .api import JsonRPCRequest
+
+
+class ResultResponse(TypedDict):
+    result: Iterable[Dict[str, Any]]
+
+
+Method = Callable[..., Union[ResultResponse, JsonRPCRequest]]
+
+
+def send_results(results: Iterable[Result]) -> ResultResponse:
+    return {'result': [result.as_dict() for result in results]}
+
+
+class Plugin:
+
+    def __init__(self, methods: Optional[list[Method]] = None) -> None:
+        self._client = JsonRPCClient()
+        self._event_handler = EventHandler()
+        self._settings: Dict[str, Any] = {}
+        if methods:
+            self.add_methods(methods)
+
+    def add_method(self, method: Method, *, name: Optional[str] = None) -> None:
+        self._event_handler.add_method(method, name=name)
+
+    def add_methods(self, methods: Iterable[Method]) -> None:
+        self._event_handler.add_methods(methods)
+
+    def on_method(self, method: Method) -> Method:
+        @wraps(method)
+        def wrapper(*args, **kwargs):
+            return method(*args, **kwargs)
+        self._event_handler.add_method(wrapper)
+        return wrapper
+
+    @property
+    def settings(self) -> Dict:
+        if self._settings is None:
+            self._settings = {}
+        self._settings = self._client.recieve().get('settings', {})
+        return self._settings
+
+    def run(self) -> None:
+        request = self._client.recieve()
+        method = request.get('method')
+        parameters = request.get('parameters', [])
+        feedback = self._event_handler(method, *parameters)
+        self._client.send(feedback)

pyflowlauncher/jsonrpc.py

@@ -0,0 +1,49 @@
+from __future__ import annotations
+from dataclasses import dataclass
+from pathlib import Path
+from typing import TYPE_CHECKING, Any, Iterable, Optional, TypedDict, Union, Dict, NotRequired
+
+
+if TYPE_CHECKING:
+    from .plugin import Method
+
+
+class JsonRPCAction(TypedDict):
+    """Flow Launcher JsonRPCAction"""
+    Method: str
+    Parameters: Iterable
+    DontHideAfterAction: NotRequired[bool]
+
+
+class Glyph(TypedDict):
+    """Flow Launcher Glyph"""
+    Glyph: str
+    FontFamily: str
+
+
+@dataclass
+class Result:
+    Title: str
+    SubTitle: Optional[str] = None
+    IcoPath: Optional[Union[str, Path]] = None
+    Score: int = 0
+    JsonRPCAction: Optional[JsonRPCAction] = None
+    ContextData: Optional[Iterable] = None
+    Glyph: Optional[Glyph] = None
+    CopyText: Optional[str] = None
+    AutoCompleteText: Optional[str] = None
+    RoundedIcon: bool = False
+
+    def as_dict(self) -> Dict[str, Any]:
+        return self.__dict__
+
+    def add_action(self, method: Method,
+                   parameters: Optional[Iterable[Any]] = None,
+                   *,
+                   dont_hide_after_action: bool = False) -> None:
+        self.JsonRPCAction = {
+            "Method": method.__name__,
+            "Parameters": parameters or [],
+            "DontHideAfterAction": dont_hide_after_action
+        }
+        self._method = method