feat: cythonize validate_flag_value, refactor (#399)

* feat: optimize ASyncGenericBase with cython

* chore: `black .`

* feat: cythonize validate_flag_value, refactor out cnegate_if_necessary, is_sync_c

* fix: AttributeError

* chore: `black .`

* fix: AttributeError

* fix: circular import

* fix: missing import

* chore: `black .`

* fix: broken import

* fix: type error

---------

Co-authored-by: github-actions[bot] <github-actions[bot]@users.noreply.github.com>

a_sync/a_sync/_flags.pxd

@@ -1 +1 @@
-cdef bint cnegate_if_necessary(str flag, object flag_value)
+cdef bint negate_if_necessary(str flag, object flag_value)

a_sync/a_sync/_flags.pyx

@@ -13,70 +13,11 @@ You can use any of the provided flags, whichever makes the most sense for your u
 :obj:`VIABLE_FLAGS`: Set of all valid flags, combining both synchronous and asynchronous indicators.
 """
 
-from typing import Any, Set
-
 from a_sync import exceptions
-
-AFFIRMATIVE_FLAGS: Set[str] = {"sync"}
-"""Set of flags indicating synchronous behavior.
-
-This set currently contains only the flag "sync", which is used to denote
-synchronous operations within the ez-a-sync library.
-
-Examples:
-    >>> 'sync' in AFFIRMATIVE_FLAGS
-    True
-
-    >>> 'async' in AFFIRMATIVE_FLAGS
-    False
-
-See Also:
-    :data:`NEGATIVE_FLAGS`: Flags indicating asynchronous behavior.
-    :data:`VIABLE_FLAGS`: All valid flags, combining both sync and async indicators.
-"""
-
-NEGATIVE_FLAGS: Set[str] = {"asynchronous"}
-"""Set of flags indicating asynchronous behavior.
-
-This set currently contains only the flag "asynchronous", which is used to denote
-asynchronous operations within the ez-a-sync library.
-
-Examples:
-    >>> 'asynchronous' in NEGATIVE_FLAGS
-    True
-
-    >>> 'sync' in NEGATIVE_FLAGS
-    False
-
-See Also:
-    :data:`AFFIRMATIVE_FLAGS`: Flags indicating synchronous behavior.
-    :data:`VIABLE_FLAGS`: All valid flags, combining both sync and async indicators.
-"""
-
-VIABLE_FLAGS: Set[str] = AFFIRMATIVE_FLAGS | NEGATIVE_FLAGS
-"""Set of all valid flags, combining both synchronous and asynchronous indicators.
-
-The ez-a-sync library uses these flags to indicate whether objects or function
-calls will be synchronous or asynchronous. You can use any of the provided flags,
-whichever makes the most sense for your use case.
-
-Examples:
-    >>> 'sync' in VIABLE_FLAGS
-    True
-
-    >>> 'asynchronous' in VIABLE_FLAGS
-    True
-
-    >>> 'invalid' in VIABLE_FLAGS
-    False
-
-See Also:
-    :data:`AFFIRMATIVE_FLAGS`: Flags indicating synchronous behavior.
-    :data:`NEGATIVE_FLAGS`: Flags indicating asynchronous behavior.
-"""
+from a_sync.a_sync.flags import AFFIRMATIVE_FLAGS, NEGATIVE_FLAGS
 
 
-def negate_if_necessary(flag: str, flag_value: bool) -> bool:
+cdef bint negate_if_necessary(str flag, object flag_value):
     """Negate the flag value if necessary based on the flag type.
 
     This function checks if the provided flag is in the set of affirmative or negative flags
@@ -102,19 +43,16 @@ def negate_if_necessary(flag: str, flag_value: bool) -> bool:
     See Also:
         - :func:`validate_flag_value`: Validates that the flag value is a boolean.
     """
-    return cnegate_if_necessary(flag, flag_value)
-
-
-cdef bint cnegate_if_necessary(str flag, object flag_value):
-    cdef bint value = validate_flag_value(flag, flag_value)
+    if not isinstance(flag_value, bool):
+        raise exceptions.InvalidFlagValue(flag, flag_value)
     if flag in AFFIRMATIVE_FLAGS:
-        return value
+        return <bint>flag_value
     elif flag in NEGATIVE_FLAGS:
-        return not value
+        return not <bint>flag_value
     raise exceptions.InvalidFlag(flag)
 
 
-def validate_flag_value(flag: str, flag_value: Any) -> bool:
+cdef bint validate_flag_value(str flag, object flag_value):
     """
     Validate that the flag value is a boolean.
 

a_sync/a_sync/_kwargs.pxd

@@ -1,2 +1,2 @@
 cdef object get_flag_name_c(dict kwargs)
-cdef bint is_sync_c(str flag, dict kwargs, bint pop_flag)
+cdef bint is_sync(str flag, dict kwargs, bint pop_flag)

a_sync/a_sync/_kwargs.pyi

@@ -23,30 +23,4 @@ def get_flag_name(kwargs: dict) -> Optional[str]:
 
         >>> get_flag_name({})
         None
-
-    See Also:
-        :func:`is_sync`: Determines if the operation should be synchronous based on the flag value.
-    """
-
-def is_sync(flag: str, kwargs: dict, pop_flag: bool = False) -> bool:
-    """
-    Determine if the operation should be synchronous based on the flag value.
-
-    Args:
-        flag (str): The name of the flag to check.
-        kwargs (dict): A dictionary of keyword arguments.
-        pop_flag (bool, optional): Whether to remove the flag from kwargs. Defaults to False.
-
-    Examples:
-        >>> is_sync('sync', {'sync': True})
-        True
-
-        >>> is_sync('asynchronous', {'asynchronous': False})
-        False
-
-        >>> is_sync('sync', {'sync': True}, pop_flag=True)
-        True
-
-    See Also:
-        :func:`get_flag_name`: Retrieves the name of the flag present in the kwargs.
     """

a_sync/a_sync/_kwargs.pyx

@@ -6,8 +6,8 @@ from typing import Optional
 from libc.stdint cimport uint8_t
 
 from a_sync import exceptions
-from a_sync.a_sync._flags import VIABLE_FLAGS
-from a_sync.a_sync._flags cimport cnegate_if_necessary
+from a_sync.a_sync._flags cimport negate_if_necessary
+from a_sync.a_sync.flags import VIABLE_FLAGS
 
 
 def get_flag_name(kwargs: dict) -> Optional[str]:
@@ -33,12 +33,10 @@ def get_flag_name(kwargs: dict) -> Optional[str]:
 
         >>> get_flag_name({})
         None
-
-    See Also:
-        :func:`is_sync`: Determines if the operation should be synchronous based on the flag value.
     """
     return get_flag_name_c(kwargs)
 
+
 cdef object get_flag_name_c(dict kwargs):
     cdef list present_flags = [flag for flag in VIABLE_FLAGS if flag in kwargs]
     cdef uint8_t flags_count = len(present_flags)
@@ -49,7 +47,7 @@ cdef object get_flag_name_c(dict kwargs):
     raise exceptions.TooManyFlags("kwargs", present_flags)
 
 
-def is_sync(flag: str, kwargs: dict, pop_flag: bool = False) -> bool:
+cdef bint is_sync(str flag, dict kwargs, bint pop_flag):
     """
     Determine if the operation should be synchronous based on the flag value.
 
@@ -58,23 +56,10 @@ def is_sync(flag: str, kwargs: dict, pop_flag: bool = False) -> bool:
         kwargs (dict): A dictionary of keyword arguments.
         pop_flag (bool, optional): Whether to remove the flag from kwargs. Defaults to False.
 
-    Examples:
-        >>> is_sync('sync', {'sync': True})
-        True
-
-        >>> is_sync('asynchronous', {'asynchronous': False})
-        False
-
-        >>> is_sync('sync', {'sync': True}, pop_flag=True)
-        True
-
     See Also:
         :func:`get_flag_name`: Retrieves the name of the flag present in the kwargs.
     """
-    return is_sync_c(flag, kwargs, pop_flag)
-
-cdef bint is_sync_c(str flag, dict kwargs, bint pop_flag):
     if pop_flag:
-        return cnegate_if_necessary(flag, kwargs.pop(flag))
+        return negate_if_necessary(flag, kwargs.pop(flag))
     else:
-        return cnegate_if_necessary(flag, kwargs[flag])
+        return negate_if_necessary(flag, kwargs[flag])

a_sync/a_sync/abstract.pyi

@@ -73,24 +73,6 @@ class ASyncABC(metaclass=ASyncMeta):
             True
         """
 
-    def __a_sync_should_await_from_kwargs__(self, kwargs: dict) -> bool:
-        """Determines execution mode from keyword arguments.
-
-        This method can be overridden to customize how flags are extracted
-        from keyword arguments.
-
-        Args:
-            kwargs: A dictionary of keyword arguments to check for flags.
-
-        Raises:
-            NoFlagsFound: If no valid flags are found in the keyword arguments.
-
-        Examples:
-            >>> instance = MyASyncClass()
-            >>> instance.__a_sync_should_await_from_kwargs__({'sync': False})
-            True
-        """
-
     @classmethod
     def __a_sync_instance_will_be_sync__(cls, args: tuple, kwargs: dict) -> bool:
         """Determines if a new instance will be synchronous.

a_sync/a_sync/abstract.pyx

@@ -13,12 +13,11 @@ import abc
 import logging
 from typing import Dict, Any, Tuple
 
-from a_sync import exceptions
 from a_sync._typing import *
-from a_sync.a_sync import modifiers
-from a_sync.a_sync cimport _flags, _kwargs
+from a_sync.a_sync cimport _kwargs
+from a_sync.a_sync._flags cimport negate_if_necessary
 from a_sync.a_sync._meta import ASyncMeta
-from a_sync.exceptions import NoFlagsFound
+
 
 logger = logging.getLogger(__name__)
 
@@ -89,7 +88,7 @@ class ASyncABC(metaclass=ASyncMeta):
 
         cdef object flag
         if flag := _kwargs.get_flag_name_c(kwargs):
-            return _kwargs.is_sync_c(<str>flag, kwargs, pop_flag=True)
+            return _kwargs.is_sync(<str>flag, kwargs, pop_flag=True)
 
         cdef ShouldAwaitCache cache
 
@@ -102,7 +101,7 @@ class ASyncABC(metaclass=ASyncMeta):
             )
 
         if not cache.is_cached:
-            cache.value = _flags.cnegate_if_necessary(
+            cache.value = negate_if_necessary(
                 self.__a_sync_flag_name__, self.__a_sync_flag_value__
             )
             cache.is_cached = True
@@ -134,7 +133,7 @@ class ASyncABC(metaclass=ASyncMeta):
             )
 
         if not cache.is_cached:
-            cache.value = _flags.cnegate_if_necessary(
+            cache.value = negate_if_necessary(
                 self.__a_sync_flag_name__, self.__a_sync_flag_value__
             )
             cache.is_cached = True
@@ -165,7 +164,7 @@ class ASyncABC(metaclass=ASyncMeta):
         cdef object flag
         cdef bint sync
         if flag := _kwargs.get_flag_name_c(kwargs):
-            sync = _kwargs.is_sync_c(<str>flag, kwargs, pop_flag=False)  # type: ignore [arg-type]
+            sync = _kwargs.is_sync(<str>flag, kwargs, pop_flag=False)  # type: ignore [arg-type]
             logger.debug(
                 "kwargs indicate the new instance created with args %s %s is %ssynchronous",
                 args,

a_sync/a_sync/base.pyi

@@ -2,11 +2,8 @@ from a_sync._typing import *
 import functools
 from _typeshed import Incomplete
 from a_sync import exceptions as exceptions
-from a_sync.a_sync._flags import (
-    VIABLE_FLAGS as VIABLE_FLAGS,
-    negate_if_necessary as negate_if_necessary,
-)
 from a_sync.a_sync.abstract import ASyncABC as ASyncABC
+from a_sync.a_sync.flags import VIABLE_FLAGS as VIABLE_FLAGS
 
 logger: Incomplete
 

a_sync/a_sync/base.pyx

@@ -5,9 +5,9 @@ from contextlib import suppress
 
 from a_sync import exceptions
 from a_sync._typing import *
-from a_sync.a_sync._flags import VIABLE_FLAGS
-from a_sync.a_sync._flags cimport cnegate_if_necessary
+from a_sync.a_sync._flags cimport negate_if_necessary
 from a_sync.a_sync.abstract import ASyncABC
+from a_sync.a_sync.flags import VIABLE_FLAGS
 
 
 logger = logging.getLogger(__name__)
@@ -118,7 +118,7 @@ class ASyncGenericBase(ASyncABC):
             except exceptions.NoFlagsFound:
                 flag = _get_a_sync_flag_name_from_class_def(cls)
                 flag_value = _get_a_sync_flag_value_from_class_def(cls, flag)
-            return cnegate_if_necessary(flag, flag_value)  # type: ignore [arg-type]
+            return negate_if_necessary(flag, flag_value)  # type: ignore [arg-type]
 
         # we need an extra var so we can log it
         cdef bint sync
@@ -130,7 +130,7 @@ class ASyncGenericBase(ASyncABC):
             flag = _get_a_sync_flag_name_from_class_def(cls)
             flag_value = _get_a_sync_flag_value_from_class_def(cls, flag)
 
-        sync = cnegate_if_necessary(flag, flag_value)  # type: ignore [arg-type]
+        sync = negate_if_necessary(flag, flag_value)  # type: ignore [arg-type]
         logger._log(
             logging.DEBUG,
             "`%s.%s` indicates default mode is %ssynchronous",
@@ -192,7 +192,7 @@ cdef str _get_a_sync_flag_name_from_signature(object cls):
     return _parse_flag_name_from_list(cls, parameters)
 
 
-cdef str _parse_flag_name_from_list(object cls, dict items):
+cdef str _parse_flag_name_from_list(object cls, object items):
     cdef list[str] present_flags
     cdef str flag
     present_flags = [flag for flag in VIABLE_FLAGS if flag in items]

a_sync/a_sync/decorator.py

@@ -1,7 +1,7 @@
 # mypy: disable-error-code=valid-type
 # mypy: disable-error-code=misc
 from a_sync._typing import *
-from a_sync.a_sync import _flags, config
+from a_sync.a_sync import config
 from a_sync.a_sync.function import (
     ASyncDecorator,
     ASyncFunction,