diff --git a/README.md b/README.md index d7b3ad4..3db24a7 100644 --- a/README.md +++ b/README.md @@ -1,7 +1,8 @@ # FletX 🚀 **The open-source GetX-inspired Python Framework for Building Reactive, Cross-Platform Apps with Flet** -[![PyPI Version](https://img.shields.io/pypi/v/fletx)](https://pypi.org/project/FletXr/) +[![PyPI Version](https://img.shields.io/pypi/v/FletXr)](https://pypi.org/project/FletXr/) +[![Downloads](https://static.pepy.tech/badge/FletXr)](https://pepy.tech/project/FletXr) [![License](https://img.shields.io/badge/license-MIT-blue)](LICENSE) [![Discord](https://img.shields.io/discord/v6trjD8m)](https://discord.gg/v6trjD8m) @@ -313,7 +314,7 @@ We welcome contributions from the community! Please see the [CONTRIBUTING.md](CO ## License 📜 -MIT © 2023 AllDotPy +MIT © 2025 AllDotPy ```bash # Happy coding! diff --git a/examples/1/main.py b/examples/1/main.py index 20553e8..2182c82 100644 --- a/examples/1/main.py +++ b/examples/1/main.py @@ -1,5 +1,6 @@ -from fletx.app import FletXApp +import asyncio import flet as ft +from fletx.app import FletXApp from routes import routes def main(page: ft.Page): diff --git a/examples/1/pages/auth/guards.py b/examples/1/pages/auth/guards.py index a9ec809..9e1784c 100644 --- a/examples/1/pages/auth/guards.py +++ b/examples/1/pages/auth/guards.py @@ -1,5 +1,5 @@ -from fletx.core.navigation.guards import RouteGuard -from fletx.core.types import RouteInfo +from fletx.core.routing.guards import RouteGuard +from fletx.core.routing.models import RouteInfo class AuthGuard(RouteGuard): diff --git a/examples/1/pages/auth/login.py b/examples/1/pages/auth/login.py index 98b2fa6..4937701 100644 --- a/examples/1/pages/auth/login.py +++ b/examples/1/pages/auth/login.py @@ -2,7 +2,7 @@ from ..shared.components import ReactivePasswordField from fletx import FletX from fletx.core.page import FletXPage -from fletx.core.router import FletXRouter +from fletx.navigation import navigate from fletx.core.state import RxBool from .controller import AuthController from .guards import AuthGuard @@ -317,6 +317,6 @@ def on_login(self, e): self.controller.login( self.email_field.value, self.password_field.value, - on_success=lambda: FletXRouter.to("/dashboard"), + on_success=lambda: navigate("/dashboard"), on_failure=lambda: print("Échec de connexion") ) \ No newline at end of file diff --git a/examples/1/pages/dashboard/settings.py b/examples/1/pages/dashboard/settings.py index 6466897..af8bf6a 100644 --- a/examples/1/pages/dashboard/settings.py +++ b/examples/1/pages/dashboard/settings.py @@ -1,7 +1,7 @@ import flet as ft from fletx.core.page import FletXPage from fletx.decorators.controllers import page_controller -from fletx.core.router import FletXRouter +from fletx.core.routing.router import FletXRouter from .controller import DashboardController @page_controller diff --git a/examples/1/routes.py b/examples/1/routes.py index 0652a77..471e91d 100644 --- a/examples/1/routes.py +++ b/examples/1/routes.py @@ -1,11 +1,59 @@ +from fletx.navigation import ( + ModuleRouter, TransitionType, RouteTransition +) +from fletx.decorators import register_router + from pages.auth.login import LoginPage from pages.dashboard.home import DashboardHomePage from pages.dashboard.settings import DashboardSettingsPage -routes = { - "/login": LoginPage, - "/dashboard": DashboardHomePage, - "/dashboard/settings": DashboardSettingsPage, -} + +routes = [ + { + 'path': '/login', + 'component': LoginPage, + 'meta':{ + 'transition': RouteTransition( + transition_type = TransitionType.ZOOM_IN, + duration = 350 + ) + } + }, + { + 'path': '/dashboard', + 'component': DashboardHomePage, + 'meta':{ + 'transition': RouteTransition( + transition_type = TransitionType.FLIP_HORIZONTAL, + duration = 350 + ) + } + }, + { + 'path': '/dashboard/settings', + 'component': DashboardSettingsPage + } +] + +@register_router +class MyAppRouter(ModuleRouter): + """My Application Routing Module.""" + + name = 'MyAppRouter' + base_path = '/' + is_root = True + routes = routes + sub_routers = [] + + + + +# MyAppRouter.add_routes(routes = routes) + +# { +# "/login": LoginPage, +# "/dashboard": DashboardHomePage, +# "/dashboard/settings": DashboardSettingsPage, +# } # DashboardHomePage().build() \ No newline at end of file diff --git a/fletx/app.py b/fletx/app.py index 9a7add6..bf63f93 100644 --- a/fletx/app.py +++ b/fletx/app.py @@ -4,14 +4,16 @@ import flet as ft from typing import Dict, Type, Optional -from fletx.core.router import FletXRouter -from fletx.core.route_config import RouteConfig +from fletx.core.routing.router import FletXRouter from fletx.core.page import FletXPage -from fletx.core.factory import FletXWidgetRegistry +# from fletx.core.factory import FletXWidgetRegistry from fletx.utils.logger import SharedLogger from fletx.utils.context import AppContext -from fletx.utils.exceptions import FletXError + +#### +## FLETX APPLICATION +##### class FletXApp: """Main application class""" @@ -32,7 +34,7 @@ def __init__( debug: Debug mode """ - self.routes = routes or {} + self.routing_module = routes or {} self.initial_route = initial_route self.theme_mode = theme_mode self.debug = debug @@ -44,8 +46,6 @@ def __init__( ) self.logger = SharedLogger.get_logger(__name__) - # Configure routes - RouteConfig.register_routes(self.routes) def run(self, **kwargs): """Deprecated method – use only in controlled environments""" @@ -73,7 +73,7 @@ def _main(self, page: ft.Page): AppContext.set_data("logger", self.logger) # FletX Router Initialization - FletXRouter.initialize(page, self.initial_route) + FletXRouter.initialize(page, initial_route = self.initial_route) self.logger.info("FletX Application initialized with success") diff --git a/fletx/cli/templates/project/pyproject.toml.tpl b/fletx/cli/templates/project/pyproject.toml.tpl index ef1b47c..622681f 100644 --- a/fletx/cli/templates/project/pyproject.toml.tpl +++ b/fletx/cli/templates/project/pyproject.toml.tpl @@ -6,6 +6,6 @@ readme = "README.md" authors = [{ name = "{{ author }}", email = "" }] requires-python = ">={{ python_version }}" dependencies = [ - "fletx", + "fletxr", "flet[all]", ] \ No newline at end of file diff --git a/fletx/cli/templates/project/requirements.txt.tpl b/fletx/cli/templates/project/requirements.txt.tpl index e8cda81..fabd7c5 100644 --- a/fletx/cli/templates/project/requirements.txt.tpl +++ b/fletx/cli/templates/project/requirements.txt.tpl @@ -1,6 +1,6 @@ # Core dependencies flet>=0.21.0 -fletx>={{ fletx_version }} +fletxr>={{ fletx_version }} # Development dependencies pytest>=7.0.0 diff --git a/fletx/core/__init__.py b/fletx/core/__init__.py index 07ce47b..4d5dfc1 100644 --- a/fletx/core/__init__.py +++ b/fletx/core/__init__.py @@ -1,16 +1,16 @@ from fletx.core.controller import FletXController from fletx.core.effects import EffectManager, Effect from fletx.core.page import FletXPage -from fletx.core.route_config import RouteConfig -from fletx.core.router import FletXRouter +from fletx.core.route_config import RouteConfig # Deprecated from fletx.core.state import ( ReactiveDependencyTracker, Observer, Reactive, Computed, RxBool, RxDict, RxInt, RxList, RxStr ) from fletx.core.types import ( - RouteInfo, BindingConfig, BindingType, + BindingConfig, BindingType, ComputedBindingConfig, FormFieldValidationRule ) +from fletx.core.router import FletXRouter from fletx.core.widget import FletXWidget __all__ = [ @@ -18,7 +18,7 @@ 'EffectManager', 'Effect', 'FletXPage', - 'RouteConfig', + 'RouteConfig', # Deprecated 'FletXRouter', 'ReactiveDependencyTracker', 'Observer', @@ -34,5 +34,4 @@ 'BindingType', 'ComputedBindingConfig', 'FormFieldValidationRule', - 'FletXWidget' ] \ No newline at end of file diff --git a/fletx/core/concurency/config.py b/fletx/core/concurency/config.py new file mode 100644 index 0000000..511997b --- /dev/null +++ b/fletx/core/concurency/config.py @@ -0,0 +1,62 @@ +from enum import Enum +from dataclasses import dataclass, field +from typing import ( + TypeVar, Generic, Callable, Any, Optional, Dict, List, + Union, Protocol, runtime_checkable +) + +# GENERIC TYPES +T = TypeVar('T') + +#### +## WORKER STATE +##### +class WorkerState(Enum): + """Possible state of a worker""" + + PENDING = "pending" + RUNNING = "running" + COMPLETED = "completed" + FAILED = "failed" + CANCELLED = "cancelled" + + +#### +## EXECUTION PRIORITY +##### +class Priority(Enum): + """Task execution priority""" + + LOW = 1 + NORMAL = 2 + HIGH = 3 + CRITICAL = 4 + + +#### +## WORKER RESULT +##### +@dataclass +class WorkerResult(Generic[T]): + """Worker execution result""" + + worker_id: str + state: WorkerState + result: Optional[T] = None + error: Optional[Exception] = None + execution_time: float = 0.0 + metadata: Dict[str, Any] = field(default_factory=dict) + + +#### +## WORKER POOL CONFIGURATION +##### +@dataclass +class WorkerPoolConfig: + """Worker Pool Configuration.""" + + max_workers: int = 4 + queue_size: int = 100 + enable_priority: bool = True + timeout: Optional[float] = None + auto_shutdown: bool = True \ No newline at end of file diff --git a/fletx/core/concurency/worker.py b/fletx/core/concurency/worker.py new file mode 100644 index 0000000..1e44c52 --- /dev/null +++ b/fletx/core/concurency/worker.py @@ -0,0 +1,638 @@ +""" +Système de Worker parallèle avec typage fort +Similar to Qt's QRunnable but more flexible and type-safe +""" + +from abc import ABC, abstractmethod +from concurrent.futures import ThreadPoolExecutor, Future, as_completed +import functools +from threading import Lock, Event +from typing import ( + TypeVar, Generic, Callable, Any, Optional, Dict, List, + Protocol, runtime_checkable +) +import logging +import traceback +from functools import wraps +import time + +from fletx.core.concurency.config import ( + WorkerState, T, Priority, WorkerResult, + WorkerPoolConfig +) + +# GENERIC TYPES +R = TypeVar('R') + + +#### +## RUNNABLE OBJECTS PROTOCOL +##### +@runtime_checkable +class Runnable(Protocol): + """Runnable objects Protocol""" + + def run(self) -> Any: + """Main execution method""" + ... + + +#### +## BASE CLASS FOR WORKERS +##### +class BaseWorker(ABC, Generic[T]): + """Base class for all workers""" + + def __init__( + self, + worker_id: Optional[str] = None, + priority: Priority = Priority.NORMAL + ): + self.worker_id = worker_id or f"worker_{id(self)}" + self.priority = priority + self.state = WorkerState.PENDING + self._result: Optional[T] = None + self._error: Optional[Exception] = None + self._execution_time: float = 0.0 + self._metadata: Dict[str, Any] = {} + self._cancelled = Event() + + @abstractmethod + def execute(self) -> T: + """ + Abstract worker logic execution method. + This method should be overridden by worker subclasses + to implement custom logic. + """ + pass + + def run(self) -> WorkerResult[T]: + """Executes the worker and return a worker result""" + + if self._cancelled.is_set(): + self.state = WorkerState.CANCELLED + return self._create_result() + + self.state = WorkerState.RUNNING + start_time = time.time() + + try: + self._result = self.execute() + self.state = WorkerState.COMPLETED + + except Exception as e: + self._error = e + self.state = WorkerState.FAILED + logging.error(f"Worker {self.worker_id} failed: {e}") + logging.debug(traceback.format_exc()) + + finally: + self._execution_time = time.time() - start_time + + return self._create_result() + + def cancel(self) -> bool: + """Cancel worker execution""" + + # Set state to cancelled + if self.state == WorkerState.PENDING: + self._cancelled.set() + self.state = WorkerState.CANCELLED + return True + return False + + def is_cancelled(self) -> bool: + """Checks if worker is cancelled""" + + return self._cancelled.is_set() + + def _create_result(self) -> WorkerResult[T]: + """Creates worker result""" + + return WorkerResult( + worker_id = self.worker_id, + state = self.state, + result = self._result, + error = self._error, + execution_time = self._execution_time, + metadata = self._metadata.copy() + ) + + +#### +## FUNCTIONS WORKER CLASS +##### +class FunctionWorker(BaseWorker[T]): + """Worker that wraps a function""" + + def __init__( + self, + func: Callable[..., T], + *args, + worker_id: Optional[str] = None, + priority: Priority = Priority.NORMAL, + **kwargs + ): + super().__init__(worker_id, priority) + self.func = func + self.args = args + self.kwargs = kwargs + + def execute(self) -> T: + """Run the function with arguments""" + + return self.func(*self.args, **self.kwargs) + + +#### +## RUNNABLE WORKER CLASS +##### +class RunnableWorker(BaseWorker[Any]): + """Worker that wraps runnable object.""" + + def __init__( + self, + runnable: Runnable, + worker_id: Optional[str] = None, + priority: Priority = Priority.NORMAL + ): + super().__init__(worker_id, priority) + self.runnable = runnable + + def execute(self) -> Any: + """Executes runnable object""" + + return self.runnable.run() + + +#### +## WORKER POOL +##### +class WorkerPool: + """Thread-safe worker pool with priority management""" + + def __init__( + self, + config: WorkerPoolConfig = WorkerPoolConfig() + ): + self.config = config + self._executor = ThreadPoolExecutor(max_workers = config.max_workers) + self._pending_workers: List[BaseWorker] = [] + self._running_futures: Dict[str, Future] = {} + self._completed_results: Dict[str, WorkerResult] = {} + self._lock = Lock() + self._shutdown = False + + def submit_worker(self, worker: BaseWorker[T]) -> str: + """Submit a worker for execution""" + + if self._shutdown: + raise RuntimeError("WorkerPool is shutdown") + + with self._lock: + + # Priority based sorted insertion + if self.config.enable_priority: + inserted = False + + # Insert the worker just before the first lower + # priority worker found in the list (worker.priority > item.prority) + for i, pending_worker in enumerate(self._pending_workers): + if worker.priority.value > pending_worker.priority.value: + self._pending_workers.insert(i, worker) + inserted = True + break + + # Append it to the end else + if not inserted: + self._pending_workers.append(worker) + + # Just append the worker to the pending list + else: + self._pending_workers.append(worker) + + self._process_pending() + return worker.worker_id + + def submit_function( + self, + func: Callable[..., T], + *args, + worker_id: Optional[str] = None, + priority: Priority = Priority.NORMAL, + **kwargs + ) -> str: + """Submit a function for execution""" + + worker = FunctionWorker( + func, *args, + worker_id = worker_id, + priority = priority, + **kwargs + ) + return self.submit_worker(worker) + + def submit_runnable( + self, + runnable: Runnable, + worker_id: Optional[str] = None, + priority: Priority = Priority.NORMAL + ) -> str: + """Submit a runnable object for execution""" + + worker = RunnableWorker( + runnable, + worker_id = worker_id, + priority = priority + ) + return self.submit_worker(worker) + + def get_result( + self, + worker_id: str, + timeout: Optional[float] = None + ) -> WorkerResult: + """return a given worker result""" + + # Is worker execution already completed ??? + with self._lock: + + # Then return the result from completed reults list + if worker_id in self._completed_results: + return self._completed_results[worker_id] + + # Wait till completion + future = self._running_futures.get(worker_id) + if future: + try: + result = future.result(timeout=timeout or self.config.timeout) + with self._lock: + self._completed_results[worker_id] = result + self._running_futures.pop(worker_id, None) + return result + + except Exception as e: + # Create error result + error_result = WorkerResult( + worker_id = worker_id, + state = WorkerState.FAILED, + error = e + ) + # And add it to completed results list + with self._lock: + self._completed_results[worker_id] = error_result + return error_result + + raise ValueError(f"Worker {worker_id} not found") + + def wait_all( + self, + timeout: Optional[float] = None + ) -> Dict[str, WorkerResult]: + """Wait for all pending workers completion.""" + + results = {} + + with self._lock: + futures = dict(self._running_futures) + + # Execute workers and store results + for worker_id, future in futures.items(): + try: + result = future.result(timeout=timeout) + results[worker_id] = result + + except Exception as e: + results[worker_id] = WorkerResult( + worker_id = worker_id, + state = WorkerState.FAILED, + error = e + ) + + return results + + def cancel_worker(self, worker_id: str) -> bool: + """Cancel a worker""" + + with self._lock: + + # Search in pending workers + for worker in self._pending_workers: + if worker.worker_id == worker_id: + worker.cancel() + self._pending_workers.remove(worker) + self._completed_results[worker_id] = worker._create_result() + return True + + # Search in pending futures + future = self._running_futures.get(worker_id) + if future: + return future.cancel() + + return False + + def get_stats(self) -> Dict[str, int]: + """Returns pool stats""" + + with self._lock: + return { + "pending": len(self._pending_workers), + "running": len(self._running_futures), + "completed": len(self._completed_results) + } + + def _process_pending(self): + """Process oending worers""" + + with self._lock: + while ( + self._pending_workers and + len(self._running_futures) < self.config.max_workers + ): + worker = self._pending_workers.pop(0) + future = self._executor.submit(worker.run) + self._running_futures[worker.worker_id] = future + + def shutdown(self, wait: bool = True): + """Shutdown the pool""" + + self._shutdown = True + self._executor.shutdown(wait=wait) + + def __enter__(self): + return self + + def __exit__(self, exc_type, exc_val, exc_tb): + if self.config.auto_shutdown: + self.shutdown() + + +_global_pool = None + +def get_global_pool() -> WorkerPool: + """Get or create the global pool""" + + global _global_pool + if _global_pool is None: + config = WorkerPoolConfig(max_workers=6, enable_priority=True) + _global_pool = WorkerPool(config) + return _global_pool + +def set_global_pool(pool: WorkerPool): + """Define the global pool""" + + global _global_pool + _global_pool = pool + + +#### +## BOUND WORKER METHOD PROXY +##### +class BoundWorkerMethod: + """ + Proxy object that binds a `WorkerTaskWrapper` to an instance (self), + allowing decorated instance methods to work seamlessly with all + background execution capabilities. + + This enables you to use: + + instance.method() + instance.method.async_call(...) + instance.method.run_and_wait(...) + + without losing access to wrapper methods. + + Attributes: + _wrapper: The original WorkerTaskWrapper + _instance: The instance to bind as the first argument + """ + + def __init__(self, wrapper: 'WorkerTaskWrapper', instance: object): + """ + Initialize the proxy with the wrapper and the instance. + + Args: + wrapper: The WorkerTaskWrapper object + instance: The object instance to bind to the call + """ + self._wrapper = wrapper + self._instance = instance + + def __call__(self, *args, **kwargs): + """ + Synchronous execution with the bound instance. + Equivalent to: wrapper(instance, *args, **kwargs) + """ + return self._wrapper(self._instance, *args, **kwargs) + + def async_call(self, *args, **kwargs) -> str: + """ + Asynchronous execution with the bound instance. + Returns a worker_id. + """ + return self._wrapper.async_call(self._instance, *args, **kwargs) + + def submit(self, *args, **kwargs) -> str: + """ + Alias for async_call(). + """ + return self._wrapper.submit(self._instance, *args, **kwargs) + + def run_and_wait(self, *args, timeout: Optional[float] = None, **kwargs): + """ + Executes the task in the background, then waits for and returns the result. + + Raises: + RuntimeError: if the task is cancelled + Exception: if the task failed with an error + """ + return self._wrapper.run_and_wait(self._instance, *args, timeout=timeout, **kwargs) + + def set_pool(self, pool: 'WorkerPool'): + """ + Sets a specific pool to use for this task. + """ + self._wrapper.set_pool(pool) + + def shutdown_default_pool(self): + """ + Shuts down the default pool used by this function, if created. + """ + self._wrapper.shutdown_default_pool() + + def __getattr__(self, name): + """ + Fallback to the wrapper’s attributes for completeness. + This makes sure any missing attributes are forwarded. + """ + return getattr(self._wrapper, name) + + +#### +## WRAPPER FOR WORKER TASK +##### +class WorkerTaskWrapper: + """ + Wrapper for @worker_task decorated functions. + It provides more flexibilities when calling a @worker_task function. + """ + + def __init__( + self, + func: Callable[..., T], + priority: Priority = Priority.NORMAL + ): + self.func = func + self.priority = priority + self._pool: Optional[WorkerPool] = None + self._default_pool: Optional[WorkerPool] = None + + # Copy original function metadata + self.__name__ = func.__name__ + self.__doc__ = func.__doc__ + self.__module__ = func.__module__ + self.__qualname__ = getattr(func, '__qualname__', func.__name__) + self.__annotations__ = getattr(func, '__annotations__', {}) + + def __call__(self, *args, **kwargs) -> T: + """direct call - executes the function synchronously""" + + return self.func(*args, **kwargs) + + def async_call(self, *args, **kwargs) -> str: + """Asynchronous call – returns a worker_id""" + + pool = self._get_pool() + return pool.submit_function( + self.func, + *args, + priority = self.priority, + **kwargs + ) + + def submit(self, *args, **kwargs) -> str: + """Alias for async_call""" + + return self.async_call(*args, **kwargs) + + def run_and_wait( + self, + *args, + timeout: Optional[float] = None, + **kwargs + ) -> T: + """Executes in parallel and waits for the result""" + + pool = self._get_pool() + worker_id = pool.submit_function( + self.func, *args, priority=self.priority, **kwargs + ) + result = pool.get_result(worker_id, timeout=timeout) + + # Failed ??? + if result.state == WorkerState.FAILED: + raise result.error + + # Or cancelled ??? + elif result.state == WorkerState.CANCELLED: + raise RuntimeError("Task was cancelled") + + return result.result + + def set_pool(self, pool: WorkerPool): + """Sets the pool to use""" + + self._pool = pool + + def _get_pool(self) -> WorkerPool: + """Gets the pool to use""" + + # 1. Explicitly defined pool + if self._pool is not None: + return self._pool + + # 2. Global pool if exists + global _global_pool + if _global_pool is not None: + return _global_pool + + # 3. Create a default pool for this function + if self._default_pool is None: + config = WorkerPoolConfig(max_workers=2, auto_shutdown=False) + self._default_pool = WorkerPool(config) + + return self._default_pool + + def shutdown_default_pool(self): + """Shuts down the default pool for this function""" + + if self._default_pool is not None: + self._default_pool.shutdown() + self._default_pool = None + + def __get__(self, instance, owner): + if instance is None: + return self + # Lie l'instance (self) à la fonction + return BoundWorkerMethod(self, instance) + + +#### WORKER TASK DECORATOR +def worker_task(priority: Priority = Priority.NORMAL): + """ + Decorator that converts a function in to a worker task + Usage: + ```python + @worker_task() + def my_function(x): + return x * 2 + + # Direct call (synchronous) + result = my_function(5) # -> 10 + + # Asynchronous call + worker_id = my_function.async_call(5) + # or + worker_id = my_function.submit(5) + + # Parallel execution with waiting + result = my_function.run_and_wait(5) + ``` + """ + + def decorator(func: Callable[[], T]) -> WorkerTaskWrapper: + return WorkerTaskWrapper(func, priority) + return decorator + +#### PARALLEL TASK DECORATOR +def parallel_task(priority: Priority = Priority.NORMAL): + """Decorator that forces parallel eecution + + Usage: + ```python + @parallel_task() + def my_fonction(x): + return x * 2 + + # This call will always be parallel and always return a worker_id + worker_id = ma_fonction(5) + ``` + """ + + def decorator(func: Callable[..., T]) -> Callable[..., str]: + wrapper = WorkerTaskWrapper(func, priority) + + @wraps(func) + def parallel_wrapper(*args, **kwargs) -> str: + return wrapper.async_call(*args, **kwargs) + + # Add utility methods + parallel_wrapper.set_pool = wrapper.set_pool + parallel_wrapper.run_and_wait = wrapper.run_and_wait + parallel_wrapper.sync_call = wrapper.__call__ + parallel_wrapper.shutdown_default_pool = wrapper.shutdown_default_pool + + return parallel_wrapper + return decorator + + \ No newline at end of file diff --git a/fletx/core/navigation/guards.py b/fletx/core/navigation/guards.py deleted file mode 100644 index 23fab05..0000000 --- a/fletx/core/navigation/guards.py +++ /dev/null @@ -1,22 +0,0 @@ -""" -Système de protection des routes (Route Guards) -""" - -from abc import ABC, abstractmethod -from typing import Any -from fletx.core.types import RouteInfo -from fletx.utils.exceptions import NavigationAborted - -class RouteGuard(ABC): - """Interface de base pour les guards""" - - @abstractmethod - def can_activate(self, route: RouteInfo) -> bool: - """Vérifie si la route peut être activée""" - pass - - @abstractmethod - def redirect(self, route: RouteInfo) -> str: - """Retourne la route de redirection si can_activate=False""" - pass - diff --git a/fletx/core/navigation/middleware.py b/fletx/core/navigation/middleware.py deleted file mode 100644 index df3e534..0000000 --- a/fletx/core/navigation/middleware.py +++ /dev/null @@ -1,34 +0,0 @@ -""" -Middleware pour intercepter la navigation -""" - -from typing import Callable, Any, Optional -from fletx.core.types import RouteInfo - -class NavigationMiddleware: - """Middleware de navigation""" - - def __init__(self): - self._before_handlers = [] - self._after_handlers = [] - - def add_before_handler(self, handler: Callable[[RouteInfo], Optional[str]]): - """Ajoute un handler avant navigation""" - self._before_handlers.append(handler) - - def add_after_handler(self, handler: Callable[[RouteInfo], None]): - """Ajoute un handler après navigation""" - self._after_handlers.append(handler) - - def run_before(self, route: RouteInfo) -> Optional[str]: - """Exécute les handlers avant navigation""" - for handler in self._before_handlers: - result = handler(route) - if result is not None: - return result - return None - - def run_after(self, route: RouteInfo): - """Exécute les handlers après navigation""" - for handler in self._after_handlers: - handler(route) \ No newline at end of file diff --git a/fletx/core/navigation/transitions.py b/fletx/core/navigation/transitions.py deleted file mode 100644 index b660221..0000000 --- a/fletx/core/navigation/transitions.py +++ /dev/null @@ -1,52 +0,0 @@ -""" -Gestion des transitions entre pages -""" - -import enum -import flet as ft -from typing import Optional, Dict, Any, Callable -from functools import partial - -class TransitionType(enum.Enum): - NATIVE = "native" - FADE = "fade" - SLIDE_LEFT = "slide_left" - SLIDE_RIGHT = "slide_right" - ZOOM = "zoom" - CUSTOM = "custom" - -class RouteTransition: - """Configuration de transition""" - - def __init__( - self, - transition_type: TransitionType = TransitionType.NATIVE, - duration: int = 300, - custom_transition: Optional[Callable] = None - ): - self.type = transition_type - self.duration = duration - self.custom = custom_transition - - def apply(self, page: ft.Page, controls: list): - """Applique la transition""" - if self.type == TransitionType.FADE: - return self._apply_fade(page, controls) - elif self.type == TransitionType.SLIDE_LEFT: - return self._apply_slide(page, controls, -1) - # ... autres transitions - - def _apply_fade(self, page: ft.Page, controls: list): - """Transition de fondu""" - for control in controls: - control.opacity = 0 - control.update() - - def animate(): - for control in controls: - control.opacity = 1 - control.update() - - page.animate_opacity(self.duration, animate) - - # ... autres méthodes de transition \ No newline at end of file diff --git a/fletx/core/observer.py b/fletx/core/observer.py deleted file mode 100644 index e69de29..0000000 diff --git a/fletx/core/page.py b/fletx/core/page.py index 1c7dea7..aa7c23a 100644 --- a/fletx/core/page.py +++ b/fletx/core/page.py @@ -10,7 +10,7 @@ ) from abc import ABC, abstractmethod from fletx.core.controller import FletXController -from fletx.core.types import RouteInfo +from fletx.core.routing.models import RouteInfo from fletx.core.di import DI from fletx.core.effects import EffectManager # from fletx.decorators import use_effect as useEffect diff --git a/fletx/core/router.py b/fletx/core/router.py index 4ffc15d..82db4e4 100644 --- a/fletx/core/router.py +++ b/fletx/core/router.py @@ -11,15 +11,22 @@ from urllib.parse import parse_qs, urlparse from typing import Dict, Any, Optional, Type, List -from fletx.core.navigation.guards import RouteGuard -from fletx.core.navigation.middleware import NavigationMiddleware -from fletx.core.navigation.transitions import RouteTransition +from fletx.core.routing.guards import RouteGuard +from fletx.core.routing.middleware import RouteMiddleware +from fletx.core.routing.transitions import RouteTransition from fletx.core.route_config import RouteConfig from fletx.core.page import FletXPage from fletx.utils.exceptions import RouteNotFoundError, NavigationError -from fletx.core.types import RouteInfo +from fletx.core.routing.models import RouteInfo from fletx.utils import get_logger +import warnings + +warnings.warn( + 'fletx.core.router.FletXRouter is deprecated and will be removed in next releases.' + 'Use fletx.routing.get_router instead.' +) + #### ## FLETX ROUTER CLASS @@ -37,7 +44,7 @@ class FletXRouter: _current_route: str = "/" _route_history: List[str] = [] _logger = get_logger('FletX.Router') - _middleware = NavigationMiddleware() + _middleware = RouteMiddleware() _guards: Dict[str, List[RouteGuard]] = {} @classmethod @@ -81,6 +88,12 @@ def to( ): """Enables complex and flexible navigation between routes.""" + warnings.warn( + 'fletx.core.router.FletXRouter.to() is deprecated ' + 'and will be removed in next releases.' + 'Use fletx.routing.navigate() instead.' + ) + MAX_REDIRECT_DEPTH = 10 try: @@ -131,7 +144,7 @@ def to( ) # 6. Middleware before navigation - redirect = cls._middleware.run_before(route_info) + redirect = cls._middleware.before_navigation(route_info,None) if redirect and redirect != clean_path: # Avoid auto-redirection return cls.to( diff --git a/fletx/core/routing/config.py b/fletx/core/routing/config.py new file mode 100644 index 0000000..2d09d04 --- /dev/null +++ b/fletx/core/routing/config.py @@ -0,0 +1,326 @@ +""" +Advanced Route Configuration for FletX + +Enhanced route configuration system supporting nested routes, +module routing, and complex route hierarchies. +""" + +from dataclasses import dataclass, field +import re +from typing import ( + Dict, List, Type, Optional, Callable, + Union, Any +) +from fletx.core.routing.models import ( + RouteType, +) +from fletx.core.routing.guards import RouteGuard +from fletx.core.routing.middleware import RouteMiddleware +from fletx.core.page import FletXPage +from fletx.utils import get_logger + + +#### +## ROUTE DEFINITION CLASS +##### +@dataclass +class RouteDefinition: + """Complete route definition with all metadata.""" + + path: str + component: Union[Type, Callable] + route_type: RouteType = RouteType.PAGE + guards: List['RouteGuard'] = field(default_factory=list) + middleware: List['RouteMiddleware'] = field(default_factory=list) + data: Dict[str, Any] = field(default_factory=dict) + children: List['RouteDefinition'] = field(default_factory=list) + parent: Optional['RouteDefinition'] = None + resolve: Dict[str, Callable] = field(default_factory=dict) # Data resolvers + meta: Dict[str, Any] = field(default_factory=dict) + + +#### +## ROUTE PATTERN +##### +class RoutePattern: + """Handles route pattern matching with parameters and wildcards.""" + + def __init__(self, pattern: str): + self.pattern = pattern + self.param_names = [] + self.regex_pattern = self._compile_pattern() + + def _compile_pattern(self) -> re.Pattern: + """Compile route pattern to regex.""" + + pattern = self.pattern + + # Handle parameters (:param) + param_pattern = r':([a-zA-Z_][a-zA-Z0-9_]*)' + matches = re.findall(param_pattern, pattern) + self.param_names = matches + + # Replace parameters with regex groups + regex_pattern = re.sub(param_pattern, r'([^/]+)', pattern) + + # Handle wildcards (*path) + regex_pattern = regex_pattern.replace('*', '(.*)') + + # Ensure exact match + regex_pattern = f'^{regex_pattern}$' + + return re.compile(regex_pattern) + + def match(self, path: str) -> Optional[Dict[str, str]]: + """Match path against pattern and extract parameters.""" + + match = self.regex_pattern.match(path) + if not match: + return None + + params = {} + for i, param_name in enumerate(self.param_names): + params[param_name] = match.group(i + 1) + + return params + + +#### +## ROUTE CONFIG +##### +class RouterConfig: + """Advanced router configuration manager.""" + + _logger = get_logger('FletX.RouterConfig') + + def __init__(self): + self._routes: Dict[str, RouteDefinition] = {} + self._route_patterns: List[tuple[RoutePattern, RouteDefinition]] = [] + self._modules: Dict[str, 'ModuleRouter'] = {} + + @classmethod + @property + def logger(cls): + if not cls._logger: + cls._logger = get_logger('FletX.RouterConfig') + return cls._logger + + def add_route( + self, + path: str, + component: Union[Type[FletXPage], Callable], + *, + route_type: RouteType = RouteType.PAGE, + guards: List[RouteGuard] = None, + middleware: List[RouteMiddleware] = None, + data: Dict = None, + children: List[RouteDefinition] = None, + resolve: Dict[str, Callable] = None, + meta: Dict = None + ) -> RouteDefinition: + """Add a route to the configuration.""" + + route_def = RouteDefinition( + path = path, + component = component, + route_type = route_type, + guards = guards or [], + middleware = middleware or [], + data = data or {}, + children = children or [], + resolve = resolve or {}, + meta = meta or {} + ) + + # Set parent-child relationships + for child in route_def.children: + child.parent = route_def + + self._routes[path] = route_def + + # Add to pattern matching if route has parameters + if ':' in path or '*' in path: + pattern = RoutePattern(path) + self._route_patterns.append((pattern, route_def)) + + self.logger.debug(f"Route added: {path} -> {component}") + return route_def + + def add_routes(self, routes: List[Dict]) -> None: + """Add multiple routes from configuration list.""" + + for route_config in routes: + self.add_route(**route_config) + + def add_nested_routes( + self, + parent_path: str, + routes: List[Dict] + ) -> None: + """Add nested routes under a parent path.""" + + parent_route = self.get_route(parent_path) + if not parent_route: + raise ValueError(f"Parent route not found: {parent_path}") + + for route_config in routes: + child_path = f"{parent_path.rstrip('/')}/{route_config['path'].lstrip('/')}" + child_route = self.add_route(child_path, **route_config) + child_route.parent = parent_route + parent_route.children.append(child_route) + + def add_module_routes( + self, + base_path: str, + module_router: 'ModuleRouter' + ) -> None: + """Add routes from a module router.""" + + self._modules[base_path] = module_router + + # Register module routes with base path prefix + for route in module_router.get_routes(): + # build full path + full_path = f"{base_path.rstrip('/')}/{route.path.lstrip('/')}" + + route.meta.update( + {'module': module_router, 'original_path': route.path} + ) + + module_route = RouteDefinition( + path = full_path, + component = route.component, + route_type = RouteType.MODULE, + guards = route.guards, + middleware = route.middleware, + data = route.data, + meta = route.meta + ) + self._routes[full_path] = module_route + + def get_route(self, path: str) -> Optional[RouteDefinition]: + """Get route definition by exact path match.""" + + return self._routes.get(path) + + def match_route(self, path: str) -> Optional[tuple[RouteDefinition, Dict[str, str]]]: + """Match path against all route patterns.""" + + # Try exact match first + exact_route = self.get_route(path) + if exact_route: + return exact_route, {} + + # Try pattern matching + for pattern, route_def in self._route_patterns: + params = pattern.match(path) + if params is not None: + return route_def, params + + return None + + def get_all_routes(self) -> Dict[str, RouteDefinition]: + """Get all registered routes.""" + + return self._routes.copy() + + def get_routes_by_type(self, route_type: RouteType) -> List[RouteDefinition]: + """Get routes filtered by type.""" + + return [ + route for route in self._routes.values() + if route.route_type == route_type + ] + + def get_child_routes(self, parent_path: str) -> List[RouteDefinition]: + """Get child routes of a parent route.""" + + parent = self.get_route(parent_path) + return parent.children if parent else [] + + def get_route_hierarchy(self, path: str) -> List[RouteDefinition]: + """Get the full hierarchy of a route (parents + self).""" + + route = self.get_route(path) + if not route: + return [] + + hierarchy = [] + current = route + while current: + hierarchy.insert(0, current) + current = current.parent + + return hierarchy + + +#### +## MODULE ROUTER (SUB-ROUTER) +##### +class ModuleRouter: + """Sub-router for handling module-specific routes.""" + + name: str = '' + base_path: str = '' + routes: List[Dict[str,Any]] = [] + sub_routers: List['ModuleRouter'] + is_root: bool = False + _config: RouterConfig = RouterConfig() + + def __init__(self): + # Add routes to the config. + self.add_routes(self.routes) + + # Add subrouters + self.add_subrouters(self.sub_routers) + + self._logger = get_logger(f'FletX.ModuleRouter.{self.name}') + + @property + def logger(self): + if not self._logger: + self._logger = get_logger(f'FletX.ModuleRouter.{self.name}') + return self._logger + + def add_route( + self, + path: str, + component: Union[Type[FletXPage], Callable], + **kwargs + ) -> RouteDefinition: + """Add route to this module.""" + + return self._config.add_route(path, component, **kwargs) + + def add_routes(self, routes: List[Dict]) -> None: + """Add multiple routes to this module.""" + + self._config.add_routes(routes) + + def get_routes(self) -> List[RouteDefinition]: + """Get all routes in this module.""" + + return list(self._config.get_all_routes().values()) + + def match_route(self, path: str) -> Optional[tuple[RouteDefinition, Dict[str, str]]]: + """Match route within this module.""" + + return self._config.match_route(path) + + def add_subrouters(self,routers: List[Type['ModuleRouter']]): + """Add Sub routers to the router""" + + for router in routers: + # Instanciate the router to recursively + # register its routes and subrouters + router_instance = router() + + # Then add it to the config + self._config.add_module_routes( + router_instance.base_path, + router_instance + ) + + +# Global router configuration instance +router_config = RouterConfig() diff --git a/fletx/core/routing/guards.py b/fletx/core/routing/guards.py new file mode 100644 index 0000000..f84c724 --- /dev/null +++ b/fletx/core/routing/guards.py @@ -0,0 +1,61 @@ +""" +Route Guard System + +This module defines a base interface for implementing route guards. +Route guards are used to control access to specific routes based on custom logic +(e.g., authentication, permissions, feature flags, etc.). +""" + +from abc import ABC, abstractmethod +from typing import Any, Optional +from fletx.core.routing.models import RouteInfo +from fletx.utils.exceptions import NavigationAborted + + +#### +## ROUTE GUARD INTERFACE +##### +class RouteGuard(ABC): + """Base interface for creating route guards. + + A route guard determines whether navigation to a route should be allowed. + It can also provide an alternative redirection route if access is denied. + """ + + @abstractmethod + def can_activate(self, route: RouteInfo) -> bool: + """ + Determines whether the given route is allowed to be activated (navigated to). + + Args: + route (RouteInfo): Information about the route being accessed. + + Returns: + bool: True if navigation to the route is allowed, False otherwise. + """ + pass + + @abstractmethod + def can_deactivate(self, current_route: RouteInfo) -> bool: + """ + Determines whether the given route is allowed to be deactivated. + Args: + current_route (RouteInfo): Information about the route being deactivated. + + Returns: + bool: _description_ + """ + pass + + @abstractmethod + def redirect_to(self, route: RouteInfo) -> Optional[str]: + """ + Specifies the fallback route to redirect to if `can_activate()` returns False. + + Args: + route (RouteInfo): Information about the route that was blocked. + + Returns: + str: A valid route path to redirect the user to (e.g., "/login"). + """ + return None diff --git a/fletx/core/routing/middleware.py b/fletx/core/routing/middleware.py new file mode 100644 index 0000000..79b580e --- /dev/null +++ b/fletx/core/routing/middleware.py @@ -0,0 +1,40 @@ +""" +Navigation Middleware + +This module provides a middleware system to intercept navigation events +and perform custom logic before and after route changes. +""" + +from typing import Callable, Any, Optional +from fletx.core.routing.models import RouteInfo +from fletx.core.routing.models import ( + NavigationIntent +) + +class RouteMiddleware: + """Navigation middleware system. + + Allows registering hooks that run before or after a route change. + Useful for logging, analytics, access control, confirmation dialogs, etc. + """ + + def before_navigation( + self, + from_route: RouteInfo, + to_route: RouteInfo + ) -> Optional[NavigationIntent]: + + """Execute before navigation. Return NavigationIntent to redirect.""" + return None + + def after_navigation(self, route_info: RouteInfo) -> None: + """Execute after successful navigation.""" + pass + + def on_navigation_error( + self, + error: Exception, + route_info: RouteInfo + ) -> None: + """Execute when navigation fails.""" + pass diff --git a/fletx/core/routing/models.py b/fletx/core/routing/models.py new file mode 100644 index 0000000..cead69d --- /dev/null +++ b/fletx/core/routing/models.py @@ -0,0 +1,141 @@ +import flet as ft +from abc import ABC, abstractmethod +from enum import Enum +from dataclasses import dataclass, field +from typing import ( + Any, Callable, Dict, List, Optional, Type, Union +) + +from fletx.core.routing.transitions import RouteTransition + + +#### +## ROUTE INFO CLASS +##### +@dataclass +class RouteInfo: + """ + Route information + Contains detailed information about a specific route, + such as its path, parameters etc... + """ + + def __init__( + self, + path: str, + params: Dict[str, Any] = None, + query: Dict[str, Any] = None, + data: Dict[str, Any] = field(default_factory=dict), + fragment: Optional[str] = None + ): + self.path = path + self.params = params or {} + self.query = query or {} + self.data = data + self.fragment = fragment + self._extra = {} + + def add_extra(self, key: str, value: Any): + """ + Adds additional data to the route + Allows associating additional data with a route, + such as metadata, security information, or context data. + """ + self._extra[key] = value + + def get_extra(self, key: str, default: Any = None) -> Any: + """ + Gets additional data + Retrieves the additional data associated with a route, + such as metadata, security information, or context data. + """ + return self._extra.get(key, default) + + @property + def full_url(self) -> str: + """Returns the complete URL including query parameters""" + + query_str = "&".join([f"{k}={v}" for k, v in self.query.items()]) + url = self.path + if query_str: + url += f"?{query_str}" + if self.fragment: + url += f"#{self.fragment}" + return url + + +#### +## NAVIGATION INTENT TYPE CLASS +##### +@dataclass +class NavigationIntent: + """Intent data for navigation with additional context.""" + + route: str + data: Dict[str, Any] = field(default_factory=dict) + replace: bool = False + clear_history: bool = False + transition: Optional['RouteTransition'] = None + + +#### +## ROUTE TYPE CLASS +##### +class RouteType(Enum): + """Types of routes supported by the router.""" + + PAGE = "page" # Full page route + VIEW = "view" # Flet view route + NESTED = "nested" # Nested route within a parent + REDIRECT = "redirect" # Redirect route + MODULE = "module" # Module route with sub-router + + +#### +## NAVIGATION MODE CLASS +##### +class NavigationMode(Enum): + """Navigation modes for handling Flet's native navigation.""" + + NATIVE = "native" # Use Flet's native Page.route + VIEWS = "views" # Use Flet's View stack + HYBRID = "hybrid" # Combine both approaches + + +##### +## ROUTER STATE +##### +@dataclass +class RouterState: + """Current state of the router.""" + + current_route: RouteInfo + history: List[RouteInfo] = field(default_factory=list) + forward_stack: List[RouteInfo] = field(default_factory=list) + navigation_mode: NavigationMode = NavigationMode.HYBRID + active_views: List[ft.View] = field(default_factory=list) + + +##### +## NAVIGATION RESULT +##### +class NavigationResult(Enum): + """Result of navigation operation.""" + + SUCCESS = "success" + BLOCKED_BY_GUARD = "blocked_by_guard" + REDIRECTED = "redirected" + ERROR = "error" + CANCELLED = "cancelled" + + +#### +## ROUTE RESOLVER INTERFACE +##### +class IRouteResolver(ABC): + """Interface for route data resolvers.""" + + @abstractmethod + def resolve(self, route_info: RouteInfo) -> Any: + """Resolve data for the route.""" + pass diff --git a/fletx/core/routing/router.py b/fletx/core/routing/router.py new file mode 100644 index 0000000..1f708be --- /dev/null +++ b/fletx/core/routing/router.py @@ -0,0 +1,547 @@ +""" +FletX Main Router System + +Advanced routing system with support for nested routes, dynamic routing, +navigation with data, history management, guards, middleware, and transitions. +Integrates with Flet's native navigation system (Page.route and Views). +""" + +import flet as ft +from typing import Dict, Any, Optional, List, Union, Callable +from urllib.parse import parse_qs, urlparse +import asyncio +from contextlib import asynccontextmanager + +from fletx.core.routing.models import ( + RouteInfo, NavigationIntent, RouterState, NavigationMode, + NavigationResult +) +from fletx.core.routing.config import ( + RouterConfig, router_config, + RouteGuard, RouteMiddleware +) +from fletx.core.page import FletXPage +from fletx.core.routing.transitions import RouteTransition +from fletx.utils.exceptions import RouteNotFoundError, NavigationError +from fletx.core.concurency.worker import ( + worker_task, parallel_task, Priority, + WorkerPool, WorkerPoolConfig +) +from fletx.utils import get_logger + + +#### +## MAIN FLETX ROUTER CLASS +##### +class FletXRouter: + """ + Advanced Router for FletX Framework + + Provides comprehensive routing with: + - Main router with sub-routers for modules + - Nested routes support + - Dynamic routing with parameters + - Navigation with data (intents) + - Navigation history management + - Route guards and middleware + - Page transitions + - Integration with Flet's native navigation + """ + + _instance: Optional['FletXRouter'] = None + _logger = get_logger('FletX.Router') + + def __init__( + self, + page: ft.Page, + config: RouterConfig = None + ): + """Initialize the router with a Flet page.""" + + self.page = page + self.config = config or router_config + self.state = RouterState( + current_route = RouteInfo(path='/'), + navigation_mode = NavigationMode.HYBRID + ) + self._resolvers: Dict[str, Callable] = {} + self._global_guards: List[RouteGuard] = [] + self._global_middleware: List[RouteMiddleware] = [] + + # Setup Flet integration + self._setup_flet_integration() + + # setup "to" method for those who still using + # an old version of fletx router + self.to = self.navigate + + @classmethod + @property + def logger(cls): + if not cls._logger: + cls._logger = get_logger('FletX.Router') + return cls._logger + + @classmethod + def get_instance(cls) -> 'FletXRouter': + """Get the singleton router instance.""" + + if cls._instance is None: + raise RuntimeError( + "Router not initialized. Call initialize() first." + ) + return cls._instance + + @classmethod + def initialize( + cls, + page: ft.Page, + initial_route: str = '/', + config: RouterConfig = None + ) -> 'FletXRouter': + """Initialize the global router instance.""" + + cls._instance = cls(page, config) + # Navigate to current root + # asyncio.run( + cls._instance.navigate(initial_route, replace = True), + # ) + + return cls._instance + + def _setup_flet_integration(self): + """Setup integration with Flet's native navigation.""" + + # Handle Flet's native route changes + self.page.on_route_change = self._on_flet_route_change + self.page.on_view_pop = self._on_flet_view_pop + + # Set initial route + initial_route = self.page.route or "/" + self.state.current_route = RouteInfo(path=initial_route) + + def _on_flet_route_change(self, e: ft.RouteChangeEvent): + """Handle Flet's native route change events.""" + + if self.state.navigation_mode in [NavigationMode.NATIVE, NavigationMode.HYBRID]: + self.logger.debug(f"Flet route changed to: {e.route}") + # Sync with our internal state + # asyncio.run( + self.navigate(e.route, sync_only=True) + # ) + + def _on_flet_view_pop(self, e: ft.ViewPopEvent): + """Handle Flet's native view pop events.""" + + if self.state.navigation_mode in [NavigationMode.VIEWS, NavigationMode.HYBRID]: + self.logger.debug("Flet view popped") + self.go_back() + + # @worker_task + def navigate( + self, + route: str, + *, + data: Dict[str, Any] = None, + replace: bool = False, + clear_history: bool = False, + transition: Optional[RouteTransition] = None, + sync_only: bool = False + ) -> NavigationResult: + """ + Navigate to a route with comprehensive options. + + Args: + route: Target route path + data: Navigation intent data + replace: Replace current route in history + clear_history: Clear navigation history + transition: Custom transition animation + sync_only: Only sync state, don't trigger navigation + """ + try: + # Parse route + parsed = urlparse(route) + path = parsed.path + query_params = parse_qs(parsed.query) + fragment = parsed.fragment + + # Create route info + route_info = RouteInfo( + path = path, + query = {k: v[0] if len(v) == 1 else v for k, v in query_params.items()}, + data = data or {}, + fragment = fragment + ) + + # Find matching route + match_result = self.config.match_route(path) + if not match_result: + raise RouteNotFoundError(f"Route not found: {path}") + + route_def, params = match_result + route_info.params = params + + # Skip navigation if sync_only + if sync_only: + self.state.current_route = route_info + return NavigationResult.SUCCESS + + # Create worker pool for concurency + with WorkerPool() as pool: + # FIrst Sublit tasks + + # It seems like our Workerpool don't work as expected 😅 + # so we cannot share it between tasks for now. I'll fix it later. + + # self._check_deactivation_guards.set_pool(pool) + # self._check_activation_guards.set_pool(pool) + # self._run_before_middleware.set_pool(pool) + # self._resolve_route_data.set_pool(pool) + # self._create_component.set_pool(pool) + # self._run_after_middleware.set_pool(pool) + + # Check deactivation guards for current route + if not self._check_deactivation_guards.run_and_wait(): + return NavigationResult.BLOCKED_BY_GUARD + + # Check activation guards for target route + guard_result = self._check_activation_guards.run_and_wait(route_info, route_def) + if guard_result != NavigationResult.SUCCESS: + return guard_result + + # Run middleware before navigation + middleware_result = self._run_before_middleware.run_and_wait( + self.state.current_route, route_info + ) + if middleware_result: + + # Middleware requested redirect + return self.navigate( + middleware_result.route, + data = middleware_result.data, + replace = middleware_result.replace, + transition = middleware_result.transition + ) + + # Resolve route data + resolved_data = self._resolve_route_data.run_and_wait(route_info, route_def) + route_info.data.update(resolved_data) + + # Update history + if clear_history: + self.state.history.clear() + self.state.forward_stack.clear() + elif not replace: + self.state.history.append(self.state.current_route) + self.state.forward_stack.clear() + + # Create and setup component + component_instance = self._create_component.run_and_wait(route_def, route_info) + self.logger.debug(f'created Component: {component_instance.__class__.__name__}') + + # Apply transition and update UI + self._apply_transition_and_update( + component_instance, + route_info, + transition or self._get_default_transition(route_def) + ) + + # Update state + self.state.current_route = route_info + + # Update Flet's native routing + if self.state.navigation_mode in [NavigationMode.NATIVE, NavigationMode.HYBRID]: + self.page.route = path + self.page.update() + + # Run middleware after navigation + self._run_after_middleware.run_and_wait(route_info) + + self.logger.info(f"Navigation successful: {path}") + return NavigationResult.SUCCESS + + except Exception as e: + self.logger.error(f"Navigation failed: {e}", exc_info=True) + self._run_error_middleware(e, route_info) + return NavigationResult.ERROR + + @parallel_task(priority = Priority.HIGH) + def navigate_with_intent(self, intent: NavigationIntent) -> NavigationResult: + """Navigate using a navigation intent object.""" + + return self.navigate( + intent.route, + data = intent.data, + replace = intent.replace, + clear_history = intent.clear_history, + transition = intent.transition + ) + + def go_back(self) -> bool: + """Navigate back in history.""" + + if not self.state.history: + self._logger.warning("No previous route in history") + return False + + previous_route = self.state.history.pop() + self.state.forward_stack.append(self.state.current_route) + + # Use async task for navigation + asyncio.run(self.navigate(previous_route.path, replace = True)) + return True + + def go_forward(self) -> bool: + """Navigate forward in history.""" + + if not self.state.forward_stack: + self._logger.warning("No forward route in history") + return False + + forward_route = self.state.forward_stack.pop() + self.state.history.append(self.state.current_route) + + # Use async task for navigation + asyncio.run(self.navigate(forward_route.path, replace = True)) + return True + + def can_go_back(self) -> bool: + """Check if can navigate back.""" + + return len(self.state.history) > 0 + + def can_go_forward(self) -> bool: + """Check if can navigate forward.""" + + return len(self.state.forward_stack) > 0 + + def get_current_route(self) -> RouteInfo: + """Get current route information.""" + + return self.state.current_route + + def get_history(self) -> List[RouteInfo]: + """Get navigation history.""" + + return self.state.history.copy() + + def add_global_guard(self, guard: RouteGuard): + """Add a global route guard.""" + + self._global_guards.append(guard) + + def add_global_middleware( + self, + middleware: RouteMiddleware + ): + """Add global middleware.""" + + self._global_middleware.append(middleware) + + def set_navigation_mode(self, mode: NavigationMode): + """Set navigation mode for Flet integration.""" + + self.state.navigation_mode = mode + self._logger.debug(f"Navigation mode set to: {mode}") + + @worker_task(priority = Priority.HIGH) + def _check_deactivation_guards(self) -> bool: + """Check if current route can be deactivated.""" + + current_route_def = self.config.get_route(self.state.current_route.path) + if not current_route_def: + return True + + # Check route-specific guards + for guard in current_route_def.guards: + if not guard.can_deactivate(self.state.current_route): + return False + + # Check global guards + for guard in self._global_guards: + if not guard.can_deactivate(self.state.current_route): + return False + + return True + + @worker_task(priority = Priority.HIGH) + def _check_activation_guards( + self, + route_info: RouteInfo, + route_def + ) -> NavigationResult: + """Check if route can be activated.""" + + all_guards = route_def.guards + self._global_guards + + for guard in all_guards: + if not guard.can_activate(route_info): + redirect_path = guard.redirect_to(route_info) + if redirect_path: + self.navigate(redirect_path, replace=True) + return NavigationResult.REDIRECTED + return NavigationResult.BLOCKED_BY_GUARD + + return NavigationResult.SUCCESS + + @worker_task(priority = Priority.HIGH) + def _run_before_middleware( + self, + from_route: RouteInfo, + to_route: RouteInfo + ) -> Optional[NavigationIntent]: + """Run before navigation middleware.""" + + all_middleware = self._global_middleware + + # Add route-specific middleware + route_def = self.config.get_route(to_route.path) + if route_def: + all_middleware.extend(route_def.middleware) + + for middleware in all_middleware: + result = middleware.before_navigation(from_route, to_route) + if result: + return result + + return None + + @worker_task(priority = Priority.HIGH) + def _run_after_middleware(self, route_info: RouteInfo): + """Run after navigation middleware.""" + + all_middleware = self._global_middleware + + route_def = self.config.get_route(route_info.path) + if route_def: + all_middleware.extend(route_def.middleware) + + for middleware in all_middleware: + middleware.after_navigation(route_info) + + @worker_task(priority = Priority.HIGH) + def _run_error_middleware( + self, + error: Exception, + route_info: RouteInfo + ): + """Run error middleware.""" + + all_middleware = self._global_middleware + + route_def = self.config.get_route(route_info.path) + if route_def: + all_middleware.extend(route_def.middleware) + + for middleware in all_middleware: + middleware.on_navigation_error(error, route_info) + + @worker_task(priority = Priority.HIGH) + def _resolve_route_data( + self, + route_info: RouteInfo, + route_def + ) -> Dict[str, Any]: + """Resolve route data using resolvers.""" + + resolved_data = {} + + for key, resolver in route_def.resolve.items(): + try: + if asyncio.iscoroutinefunction(resolver): + resolved_data[key] = resolver(route_info) + else: + resolved_data[key] = resolver(route_info) + except Exception as e: + self._logger.error(f"Data resolver failed for {key}: {e}") + + return resolved_data + + @worker_task(priority = Priority.HIGH) + def _create_component( + self, + route_def, + route_info: RouteInfo + ): + """Create and initialize the route component.""" + + component_class = route_def.component + + if ( + isinstance(component_class, type) + and + issubclass(component_class, FletXPage) + ): + # FletX page + instance = component_class() + if hasattr(instance, 'route_info'): + instance.route_info = route_info + return instance + + elif callable(component_class): + # Callable component + return component_class(route_info) + + else: + raise NavigationError(f"Invalid component type: {type(component_class)}") + + @worker_task(priority = Priority.HIGH) + def _apply_transition_and_update( + self, + component, + route_info: RouteInfo, + transition: Optional[RouteTransition] + ): + """Apply transition and update the UI.""" + + if hasattr(component, 'build'): + content = component.build() + else: + content = component + + # Handle different navigation modes + if self.state.navigation_mode == NavigationMode.VIEWS: + + # Use Flet Views + view = ft.View( + route = route_info.path, + controls=[content] if not isinstance(content, list) else content + ) + self.page.views.append(view) + self.state.active_views.append(view) + + else: + # Direct page update + if transition: + content = asyncio.run( + transition.apply( + self.page, + [content] if not isinstance(content, list) else content + ) + ) + + self.page.clean() + if isinstance(content, list): + self.page.add(*content) + else: + self.page.add(content) + + self.page.update() + + # Call lifecycle methods + if hasattr(component, 'did_mount'): + try: + if asyncio.iscoroutinefunction(component.did_mount): + component.did_mount() + else: + component.did_mount() + except Exception as e: + self._logger.error(f"did_mount() failed: {e}") + + def _get_default_transition(self, route_def) -> Optional[RouteTransition]: + """Get default transition for route.""" + + if 'transition' in route_def.meta: + return route_def.meta['transition'] + return None diff --git a/fletx/core/routing/transitions.py b/fletx/core/routing/transitions.py new file mode 100644 index 0000000..7488b8b --- /dev/null +++ b/fletx/core/routing/transitions.py @@ -0,0 +1,618 @@ +""" +Page Transition Management + +This module defines various transition types and logic for animating +UI changes when navigating between routes in a Flet app. +""" + +import asyncio +import enum +import flet as ft +from typing import List, Optional, Dict, Any, Callable +from functools import partial + +from fletx.core.concurency.worker import ( + worker_task, parallel_task, Priority, + WorkerPool, WorkerPoolConfig +) +from fletx.utils import get_logger + + +#### +## TRANSITION TYPE +##### +class TransitionType(enum.Enum): + """Supported transition types.""" + + NONE = "none" + FADE = "fade" + SLIDE_LEFT = "slide_left" + SLIDE_RIGHT = "slide_right" + SLIDE_UP = "slide_up" + SLIDE_DOWN = "slide_down" + ZOOM_IN = "zoom_in" + ZOOM_OUT = "zoom_out" + FLIP_HORIZONTAL = "flip_horizontal" + FLIP_VERTICAL = "flip_vertical" + ROTATE = "rotate" + PUSH_LEFT = "push_left" + PUSH_RIGHT = "push_right" + PUSH_UP = "push_up" + PUSH_DOWN = "push_down" + SCALE = "scale" + CUSTOM = "custom" + + +#### +## TRANSITION DIRECTION +##### +class TransitionDirection(enum.Enum): + """Direction for directional transitions.""" + + LEFT = "left" + RIGHT = "right" + UP = "up" + DOWN = "down" + + +#### +## EASING FUNCTION CHOICES +##### +class EasingFunction(enum.Enum): + """Easing functions for smooth animations.""" + + LINEAR = "linear" + EASE_IN = "easeIn" + EASE_OUT = "easeOut" + EASE_IN_OUT = "easeInOut" + BOUNCE_IN = "bounceIn" + BOUNCE_OUT = "bounceOut" + ELASTIC_IN = "elasticIn" + ELASTIC_OUT = "elasticOut" + CUBIC_BEZIER = "cubicBezier" + + +#### +## ROUTE TRANSITION +##### +class RouteTransition: + """ + Route transition configuration and execution. + + Handles page transitions with various animation types, durations, + and easing functions for smooth navigation experiences. + """ + + def __init__( + self, + transition_type: TransitionType = TransitionType.FADE, + duration: int = 300, + easing: EasingFunction = EasingFunction.EASE_IN_OUT, + direction: Optional[TransitionDirection] = None, + custom_transition: Optional[Callable] = None, + reverse_on_back: bool = True, + **kwargs + ): + """ + Initialize route transition. + + Args: + transition_type: Type of transition animation + duration: Duration in milliseconds + easing: Easing function for smooth animation + direction: Direction for directional transitions + custom_transition: Custom transition function + reverse_on_back: Reverse transition when going back + **kwargs: Additional transition parameters + """ + self.type = transition_type + self.duration = duration + self.easing = easing + self.direction = direction + self.custom = custom_transition + self.reverse_on_back = reverse_on_back + self.params = kwargs + self._logger = get_logger('FletX.RouteTransition') + self._animation_complete = False + self._current_animation = None + + def _get_animation_curve(self) -> str: + """Convert EasingFunction to Flet animation curve.""" + return self.easing.value + + def _create_animation(self, duration: Optional[int] = None) -> ft.Animation: + """Create Flet animation object with proper configuration.""" + return ft.Animation( + duration=duration or self.duration, + curve=self._get_animation_curve() + ) + + @worker_task(priority=Priority.HIGH) + async def apply( + self, + page: ft.Page, + new_controls: List[ft.Control], + old_controls: List[ft.Control] = None, + is_back_navigation: bool = False + ) -> List[ft.Control]: + """ + Apply transition to controls. + + Args: + page: Flet page object + new_controls: New controls to transition in + old_controls: Old controls to transition out + is_back_navigation: Whether this is back navigation + + Returns: + List of controls after transition + """ + try: + if self.type == TransitionType.NONE: + return new_controls + + # Determine actual transition type (reverse if back navigation) + actual_type = self._get_actual_transition_type(is_back_navigation) + + # Apply the transition + if actual_type == TransitionType.FADE: + return await self._apply_fade(page, new_controls, old_controls) + + # Slides + elif actual_type in [ + TransitionType.SLIDE_LEFT, TransitionType.SLIDE_RIGHT, + TransitionType.SLIDE_UP, TransitionType.SLIDE_DOWN + ]: + return await self._apply_slide(page, new_controls, old_controls, actual_type) + + # Zoom + elif actual_type in [TransitionType.ZOOM_IN, TransitionType.ZOOM_OUT]: + return await self._apply_zoom(page, new_controls, old_controls, actual_type) + + # Pushes + elif actual_type in [ + TransitionType.PUSH_LEFT, TransitionType.PUSH_RIGHT, + TransitionType.PUSH_UP, TransitionType.PUSH_DOWN + ]: + return await self._apply_push(page, new_controls, old_controls, actual_type) + + # Scale + elif actual_type == TransitionType.SCALE: + return await self._apply_scale(page, new_controls, old_controls) + + # Flip + elif actual_type in [TransitionType.FLIP_HORIZONTAL, TransitionType.FLIP_VERTICAL]: + return await self._apply_flip(page, new_controls, old_controls, actual_type) + + # Rotate + elif actual_type == TransitionType.ROTATE: + return await self._apply_rotate(page, new_controls, old_controls) + + # Custom Transition + elif actual_type == TransitionType.CUSTOM and self.custom: + return await self._apply_custom(page, new_controls, old_controls) + + else: + self._logger.warning(f"Unsupported transition type: {actual_type}") + return new_controls + + except Exception as e: + self._logger.error(f"Transition failed: {e}") + return new_controls + + def _get_actual_transition_type(self, is_back_navigation: bool) -> TransitionType: + """Get the actual transition type, considering back navigation.""" + + if not is_back_navigation or not self.reverse_on_back: + return self.type + + # Reverse transitions for back navigation + reverse_map = { + TransitionType.SLIDE_LEFT: TransitionType.SLIDE_RIGHT, + TransitionType.SLIDE_RIGHT: TransitionType.SLIDE_LEFT, + TransitionType.SLIDE_UP: TransitionType.SLIDE_DOWN, + TransitionType.SLIDE_DOWN: TransitionType.SLIDE_UP, + TransitionType.ZOOM_IN: TransitionType.ZOOM_OUT, + TransitionType.ZOOM_OUT: TransitionType.ZOOM_IN, + TransitionType.PUSH_LEFT: TransitionType.PUSH_RIGHT, + TransitionType.PUSH_RIGHT: TransitionType.PUSH_LEFT, + TransitionType.PUSH_UP: TransitionType.PUSH_DOWN, + TransitionType.PUSH_DOWN: TransitionType.PUSH_UP, + } + + return reverse_map.get(self.type, self.type) + + async def _apply_fade( + self, + page: ft.Page, + new_controls: List[ft.Control], + old_controls: List[ft.Control] = None + ) -> List[ft.Control]: + """Apply fade transition using Flet's animate_opacity.""" + + # Create container for new controls with fade animation + new_container = ft.Container( + content = ft.Column(new_controls, tight=True, expand=True), + opacity = 0, + animate_opacity = self._create_animation(), + expand = True + ) + + if old_controls: + # Create container for old controls + old_container = ft.Container( + content = ft.Column(old_controls, tight=True, expand=True), + opacity = 1, + animate_opacity = self._create_animation(self.duration // 2), + expand = True + ) + + # Use Stack to overlay containers + stack = ft.Stack([old_container, new_container], expand=True) + page.clean() + page.add(stack) + page.update() + + # Start fade out of old content + await asyncio.sleep(0.01) + old_container.opacity = 0 + page.update() + + # Wait for half duration, then fade in new content + await asyncio.sleep((self.duration // 2) / 1000) + new_container.opacity = 1 + page.update() + + # Wait for animation to complete + await asyncio.sleep((self.duration // 2) / 1000) + else: + # Just fade in new content + page.clean() + page.add(new_container) + page.update() + + await asyncio.sleep(0.01) + new_container.opacity = 1 + page.update() + await asyncio.sleep(self.duration / 1000) + + # Replace with final controls + page.clean() + page.add(*new_controls) + page.update() + + return new_controls + + async def _apply_slide( + self, + page: ft.Page, + new_controls: List[ft.Control], + old_controls: List[ft.Control], + slide_type: TransitionType + ) -> List[ft.Control]: + """Apply slide transition using Flet's animate_offset.""" + + # Determine slide direction offsets + direction_map = { + TransitionType.SLIDE_LEFT: (ft.Offset(1, 0), ft.Offset(-1, 0)), + TransitionType.SLIDE_RIGHT: (ft.Offset(-1, 0), ft.Offset(1, 0)), + TransitionType.SLIDE_UP: (ft.Offset(0, 1), ft.Offset(0, -1)), + TransitionType.SLIDE_DOWN: (ft.Offset(0, -1), ft.Offset(0, 1)) + } + + new_start_offset, old_end_offset = direction_map[slide_type] + + # Create animated containers + new_container = ft.Container( + content = ft.Column(new_controls, tight=True, expand=True), + offset = new_start_offset, + animate_offset = self._create_animation(), + expand = True + ) + + if old_controls: + old_container = ft.Container( + content = ft.Column(old_controls, tight=True, expand=True), + offset = ft.Offset(0, 0), + animate_offset = self._create_animation(), + expand = True + ) + + # Place both containers in a stack + stack = ft.Stack([old_container, new_container], expand=True) + page.clean() + page.add(stack) + page.update() + + # Start slide animation + await asyncio.sleep(0.01) + old_container.offset = old_end_offset + new_container.offset = ft.Offset(0, 0) + page.update() + + # Wait for animation to complete + await asyncio.sleep(self.duration / 1000) + else: + page.clean() + page.add(new_container) + page.update() + + await asyncio.sleep(0.01) + new_container.offset = ft.Offset(0, 0) + page.update() + await asyncio.sleep(self.duration / 1000) + + # Replace with final controls + page.clean() + page.add(*new_controls) + page.update() + + return new_controls + + async def _apply_zoom( + self, + page: ft.Page, + new_controls: List[ft.Control], + old_controls: List[ft.Control], + zoom_type: TransitionType + ) -> List[ft.Control]: + """Apply zoom transition using Flet's animate_scale.""" + + initial_scale = 0.0 if zoom_type == TransitionType.ZOOM_IN else 1.5 + + # Create animated container with scale + new_container = ft.Container( + content = ft.Column(new_controls, tight=True, expand=True), + scale = ft.Scale(initial_scale), + animate_scale = self._create_animation(), + expand = True + ) + + if old_controls: + # Create old container with fade out + old_container = ft.Container( + content = ft.Column(old_controls, tight=True, expand=True), + opacity = 1, + animate_opacity = self._create_animation(self.duration // 2), + expand = True + ) + + stack = ft.Stack([old_container, new_container], expand=True) + page.clean() + page.add(stack) + page.update() + + # Start animations + await asyncio.sleep(0.01) + old_container.opacity = 0 + new_container.scale = ft.Scale(1.0) + page.update() + else: + page.clean() + page.add(new_container) + page.update() + + # Start zoom animation + await asyncio.sleep(0.01) + new_container.scale = ft.Scale(1.0) + page.update() + + # Wait for animation + await asyncio.sleep(self.duration / 1000) + + # Replace with final controls + page.clean() + page.add(*new_controls) + page.update() + + return new_controls + + async def _apply_scale( + self, + page: ft.Page, + new_controls: List[ft.Control], + old_controls: List[ft.Control] + ) -> List[ft.Control]: + """Apply scale transition (similar to zoom but with different behavior).""" + + return await self._apply_zoom(page, new_controls, old_controls, TransitionType.ZOOM_IN) + + async def _apply_rotate( + self, + page: ft.Page, + new_controls: List[ft.Control], + old_controls: List[ft.Control] + ) -> List[ft.Control]: + """Apply rotation transition using Flet's animate_rotation.""" + + from math import pi + + # Create animated container with rotation + new_container = ft.Container( + content = ft.Column(new_controls, tight=True, expand=True), + rotate = ft.Rotate(pi/2, alignment=ft.alignment.center), + animate_rotation = self._create_animation(), + opacity = 0, + animate_opacity = self._create_animation(self.duration // 2), + expand = True + ) + + if old_controls: + old_container = ft.Container( + content = ft.Column(old_controls, tight=True, expand=True), + rotate = ft.Rotate(0, alignment=ft.alignment.center), + animate_rotation = self._create_animation(), + opacity = 1, + animate_opacity = self._create_animation(self.duration // 2), + expand = True + ) + + stack = ft.Stack([old_container, new_container], expand=True) + page.clean() + page.add(stack) + page.update() + + # Start rotation and fade + await asyncio.sleep(0.01) + old_container.rotate = ft.Rotate(-pi/2, alignment=ft.alignment.center) + old_container.opacity = 0 + new_container.rotate = ft.Rotate(0, alignment=ft.alignment.center) + new_container.opacity = 1 + page.update() + else: + page.clean() + page.add(new_container) + page.update() + + await asyncio.sleep(0.01) + new_container.rotate = ft.Rotate(0, alignment=ft.alignment.center) + new_container.opacity = 1 + page.update() + + # Wait for animation + await asyncio.sleep(self.duration / 1000) + + # Replace with final controls + page.clean() + page.add(*new_controls) + page.update() + + return new_controls + + async def _apply_flip( + self, + page: ft.Page, + new_controls: List[ft.Control], + old_controls: List[ft.Control], + flip_type: TransitionType + ) -> List[ft.Control]: + """Apply flip transition using scale animation to simulate 3D flip.""" + + # Use scale to simulate flip effect + if flip_type == TransitionType.FLIP_HORIZONTAL: + # Horizontal flip uses X-axis scale + mid_scale = ft.Scale(scale_x=0, scale_y=1) + else: + # Vertical flip uses Y-axis scale + mid_scale = ft.Scale(scale_x=1, scale_y=0) + + if old_controls: + # First half: scale old content to 0 + old_container = ft.Container( + content=ft.Column(old_controls, tight=True, expand=True), + scale=ft.Scale(1), + animate_scale=self._create_animation(self.duration // 2), + expand=True + ) + + page.clean() + page.add(old_container) + page.update() + + await asyncio.sleep(0.01) + old_container.scale = mid_scale + page.update() + await asyncio.sleep((self.duration // 2) / 1000) + + # Second half: scale new content from 0 to 1 + new_container = ft.Container( + content=ft.Column(new_controls, tight=True, expand=True), + scale=mid_scale, + animate_scale=self._create_animation(self.duration // 2), + expand=True + ) + + page.clean() + page.add(new_container) + page.update() + + await asyncio.sleep(0.01) + new_container.scale = ft.Scale(1) + page.update() + await asyncio.sleep((self.duration // 2) / 1000) + + # Replace with final controls + page.clean() + page.add(*new_controls) + page.update() + + return new_controls + + async def _apply_push( + self, + page: ft.Page, + new_controls: List[ft.Control], + old_controls: List[ft.Control], + push_type: TransitionType + ) -> List[ft.Control]: + """Apply push transition (similar to slide but both move together).""" + + # Push is similar to slide but both containers move + return await self._apply_slide( + page, new_controls, old_controls, + self._push_to_slide_type(push_type) + ) + + def _push_to_slide_type(self, push_type: TransitionType) -> TransitionType: + """Convert push type to equivalent slide type.""" + push_to_slide = { + TransitionType.PUSH_LEFT: TransitionType.SLIDE_LEFT, + TransitionType.PUSH_RIGHT: TransitionType.SLIDE_RIGHT, + TransitionType.PUSH_UP: TransitionType.SLIDE_UP, + TransitionType.PUSH_DOWN: TransitionType.SLIDE_DOWN, + } + return push_to_slide.get(push_type, TransitionType.SLIDE_LEFT) + + async def _apply_custom( + self, + page: ft.Page, + new_controls: List[ft.Control], + old_controls: List[ft.Control] + ) -> List[ft.Control]: + """Apply custom transition function.""" + + if self.custom: + try: + return await self.custom(page, new_controls, old_controls, self.duration) + except Exception as e: + self._logger.error(f"Custom transition failed: {e}") + + # Fallback to fade if custom fails + return await self._apply_fade(page, new_controls, old_controls) + + def set_animation_end_callback(self, callback: Callable): + """Set callback to be called when animation ends.""" + self._animation_end_callback = callback + + async def wait_for_completion(self): + """Wait for the current animation to complete.""" + if self._current_animation: + await asyncio.sleep(self.duration / 1000) + self._animation_complete = True + + +# Utility functions for creating common transitions +def create_slide_transition( + direction: TransitionDirection, + duration: int = 300 +) -> RouteTransition: + """Create a slide transition in the specified direction.""" + + transition_map = { + TransitionDirection.LEFT: TransitionType.SLIDE_LEFT, + TransitionDirection.RIGHT: TransitionType.SLIDE_RIGHT, + TransitionDirection.UP: TransitionType.SLIDE_UP, + TransitionDirection.DOWN: TransitionType.SLIDE_DOWN, + } + return RouteTransition(transition_map[direction], duration) + +def create_fade_transition(duration: int = 300) -> RouteTransition: + """Create a simple fade transition.""" + + return RouteTransition(TransitionType.FADE, duration) + +def create_zoom_transition(zoom_in: bool = True, duration: int = 300) -> RouteTransition: + """Create a zoom transition.""" + + transition_type = TransitionType.ZOOM_IN if zoom_in else TransitionType.ZOOM_OUT + return RouteTransition(transition_type, duration) \ No newline at end of file diff --git a/fletx/core/types.py b/fletx/core/types.py index 8d8c3a8..213a28b 100644 --- a/fletx/core/types.py +++ b/fletx/core/types.py @@ -1,53 +1,19 @@ """ -FletX Types module +FletX Core Types and Interfaces. + +This module defines the core types, interfaces, and data structures +used throughout the FletX system. """ +import flet as ft +from abc import ABC, abstractmethod from typing import ( Dict, Any, Type, Optional, Callable, List, Union ) -from dataclasses import dataclass +from dataclasses import dataclass, field from enum import Enum -#### -## ROUTE INFO CLASS -##### -@dataclass -class RouteInfo: - """ - Route information - Contains detailed information about a specific route, - such as its path, parameters etc... - """ - - def __init__( - self, - path: str, - params: Dict[str, Any] = None, - query: Dict[str, Any] = None - ): - self.path = path - self.params = params or {} - self.query = query or {} - self._extra = {} - - def add_extra(self, key: str, value: Any): - """ - Adds additional data to the route - Allows associating additional data with a route, - such as metadata, security information, or context data. - """ - self._extra[key] = value - - def get_extra(self, key: str, default: Any = None) -> Any: - """ - Gets additional data - Retrieves the additional data associated with a route, - such as metadata, security information, or context data. - """ - return self._extra.get(key, default) - - #### ## BINDING TYPE CLASS ##### diff --git a/fletx/decorators/__init__.py b/fletx/decorators/__init__.py index 0def2d9..2fb9de5 100644 --- a/fletx/decorators/__init__.py +++ b/fletx/decorators/__init__.py @@ -11,9 +11,9 @@ reactive_when, reactive_computed ) from fletx.decorators.controllers import page_controller, with_controller -from fletx.decorators.route import register_route - +from fletx.decorators.route import register_router from fletx.decorators.effects import use_effect +from fletx.core.concurency.worker import worker_task, parallel_task __all__ = [ # Widget Reactivity @@ -41,10 +41,14 @@ "with_controller", # Routing - "register_route", + "register_router", # Effects "use_effect", # "effect", # "use_memo", + + # Background + 'worker_task', + 'parallel_task' ] \ No newline at end of file diff --git a/fletx/decorators/route.py b/fletx/decorators/route.py index 754a302..f15b9ab 100644 --- a/fletx/decorators/route.py +++ b/fletx/decorators/route.py @@ -7,15 +7,19 @@ """ from typing import Type, Callable -from fletx.core.route_config import RouteConfig -from fletx.core.page import FletXPage +from fletx.core.routing.config import ( + router_config, ModuleRouter +) -#### REGISTER ROUTE -def register_route(path: str): - """Decorator to automatically register a route""" - - def decorator(page_class: Type[FletXPage]): - RouteConfig.register_route(path, page_class) - return page_class - return decorator \ No newline at end of file +#### REGISTER ROUTER +def register_router(cls: ModuleRouter): + """Decorator that automatically registers module routes""" + # Initialisation du routeur parent + router = cls() + + # Enregistrement dans la config globale + if cls.is_root: + router_config.add_module_routes('', router) + + return cls \ No newline at end of file diff --git a/fletx/navigation/__init__.py b/fletx/navigation/__init__.py new file mode 100644 index 0000000..de2dac9 --- /dev/null +++ b/fletx/navigation/__init__.py @@ -0,0 +1,66 @@ +import asyncio + +from fletx.core.routing.router import ( + FletXRouter +) +from fletx.core.routing.config import ( + RoutePattern, RouterConfig, router_config, + ModuleRouter +) +from fletx.core.routing.guards import RouteGuard +from fletx.core.routing.middleware import RouteMiddleware +from fletx.core.routing.transitions import ( + TransitionType, RouteTransition +) +from fletx.core.routing.models import ( + RouteInfo, RouterState, RouteType, + NavigationIntent, NavigationMode, + NavigationResult, IRouteResolver +) +# from fletx.core.background import run_background + + +# Convenience functions for global router access + +def get_router() -> FletXRouter: + """Get the global router instance.""" + return FletXRouter.get_instance() + +def navigate(route: str, **kwargs) -> NavigationResult: + """Navigate using the global router.""" + router = get_router() + return router.navigate(route, **kwargs) + +def go_back() -> bool: + """Go back using the global router.""" + return get_router().go_back() + +def go_forward() -> bool: + """Go forward using the global router.""" + return get_router().go_forward() + + +__all__ = [ + 'RouteGuard', + 'RouteMiddleware', + 'TransitionType', + 'RouteTransition', + 'RoutePattern', + 'RouterConfig', + 'FletXRouter', + 'NavigationResult', + 'RouteInfo', + 'RouterState', + 'RouteType', + 'NavigationIntent', + 'NavigationMode', + 'IRouteResolver', + 'ModuleRouter', + 'router_config', + + # FUNCTIONS + 'get_router', + 'navigate', + 'go_back', + 'go_forward' +] \ No newline at end of file diff --git a/fletx/utils/logger.py b/fletx/utils/logger.py index ad4bd69..da24748 100644 --- a/fletx/utils/logger.py +++ b/fletx/utils/logger.py @@ -2,6 +2,7 @@ Logging system for FletX """ +import os import logging import sys import threading @@ -16,6 +17,7 @@ class SharedLogger: _logger: Optional[logging.Logger] = None _lock = threading.Lock() + debug_mode = os.getenv('FLETX_DEBUG','0') == 1 @classmethod def get_logger(cls, name: str = "FletX") -> logging.Logger: @@ -51,20 +53,25 @@ def _initialize_logger(cls, name: str,debug: bool = False): def debug(self, message: str): """Log a debug message""" - self.logger.debug(message) + if self.debug_mode: + self.logger.debug(message) def info(self, message: str): """Log an info level message""" - self.logger.info(message) + if self.debug_mode: + self.logger.info(message) def warning(self, message: str): """Log a warnning level mesage""" - self.logger.warning(message) + if self.debug_mode: + self.logger.warning(message) def error(self, message: str,* args, **kwargs): """Log an error level message""" - self.logger.error(message) + if self.debug_mode: + self.logger.error(message) def critical(self, message: str): """Log a critical level message""" - self.logger.critical(message) + if self.debug_mode: + self.logger.critical(message)