cowbird.utils

Attributes

`CONTENT_TYPE_ANY`
`CONTENT_TYPE_JSON`
`CONTENT_TYPE_FORM`
`CONTENT_TYPE_HTML`
`CONTENT_TYPE_PLAIN`
`CONTENT_TYPE_APP_XML`
`CONTENT_TYPE_TXT_XML`
`FORMAT_TYPE_MAPPING`
`SUPPORTED_ACCEPT_TYPES`
`SUPPORTED_FORMAT_TYPES`
`KNOWN_CONTENT_TYPES`
`CLI_MODE_CFG`
`USE_TEST_CELERY_APP_CFG`
`EnumClassType`
`LOGGER`
`SingletonMetaType`
`null`

Classes

`ExtendedEnum`	Utility `enum.Enum` methods.
`SingletonMeta`	A metaclass that creates a Singleton base class when called.
`NullType`	Represents a null value to differentiate from None.

Functions

`get_logger`(→ logging.Logger)	Immediately sets the logger level to avoid duplicate log outputs from the root logger and this logger when
`set_logger_config`(→ logging.Logger)	Applies the provided logging configuration settings to the logger.
`print_log`(→ None)	Logs the requested message to the logger and optionally enforce printing to the console according to configuration
`raise_log`(→ NoReturn)	Logs the provided message to the logger and raises the corresponding exception afterwards.
`bool2str`(→ str)	Converts `value` to explicit `"true"` or `"false"` `str` with permissive variants comparison
`islambda`(→ bool)	Evaluate if argument is a callable `lambda` expression.
`configure_celery`(→ None)
`get_app_config`(→ pyramid.config.Configurator)	Generates application configuration with all required utilities and settings configured.
`get_settings_from_config_ini`(...)	Loads configuration INI settings with additional handling.
`get_registry`(→ Optional[pyramid.registry.Registry])	Retrieves the application `registry` from various containers referencing to it.
`get_json`(→ cowbird.typedefs.JSON)	Retrieves the 'JSON' body of a response using the property/callable according to the response's implementation.
`get_header`(→ Optional[str])	Retrieves `header_name` by fuzzy match (independently of upper/lower-case and underscore/dash) from various
`convert_response`(→ pyramid.response.Response)	Converts a `requests.Response` object to an equivalent `pyramid.response.Response` object.
`get_settings`(→ cowbird.typedefs.SettingsType)	Retrieve application settings from a supported container.
`fully_qualified_name`(→ str)	Obtains the `'<module>.<name>'` full path definition of the object to allow finding and importing it.
`log_request_format`(→ str)
`log_request`(→ None)	Subscriber event that logs basic details about the incoming requests.
`log_exception_tween`(...)	Tween factory that logs any exception before re-raising it.
`is_json_body`(→ bool)
`is_null`(→ bool)
`get_config_path`(→ str)
`get_ssl_verify`(→ bool)
`get_timeout`(→ int)
`apply_new_path_permissions`(→ None)	Applies new permissions to a path, if required.
`update_filesystem_permissions`(→ int)	Applies/remove read, write and execute permissions (on `others` only) to the input file system permissions.
`apply_default_path_ownership`(→ None)	Applies default ownership to a path, if required.

Module Contents

cowbird.utils.CONTENT_TYPE_ANY = '*/*'[source]

cowbird.utils.CONTENT_TYPE_JSON = 'application/json'[source]

cowbird.utils.CONTENT_TYPE_FORM = 'application/x-www-form-urlencoded'[source]

cowbird.utils.CONTENT_TYPE_HTML = 'text/html'[source]

cowbird.utils.CONTENT_TYPE_PLAIN = 'text/plain'[source]

cowbird.utils.CONTENT_TYPE_APP_XML = 'application/xml'[source]

cowbird.utils.CONTENT_TYPE_TXT_XML = 'text/xml'[source]

cowbird.utils.FORMAT_TYPE_MAPPING[source]

cowbird.utils.SUPPORTED_ACCEPT_TYPES[source]

cowbird.utils.SUPPORTED_FORMAT_TYPES[source]

cowbird.utils.KNOWN_CONTENT_TYPES[source]

cowbird.utils.CLI_MODE_CFG = 'cli_mode'[source]

cowbird.utils.USE_TEST_CELERY_APP_CFG = 'use_test_celery_app'[source]

cowbird.utils.EnumClassType[source]

cowbird.utils.get_logger(name: str, level: int | None = None, force_stdout: bool = None, message_format: str | None = None, datetime_format: str | None = None) → logging.Logger[source]: Immediately sets the logger level to avoid duplicate log outputs from the root logger and this logger when level is logging.NOTSET.

cowbird.utils.LOGGER[source]

cowbird.utils.set_logger_config(logger: logging.Logger, force_stdout: bool = False, message_format: str | None = None, datetime_format: str | None = None) → logging.Logger[source]: Applies the provided logging configuration settings to the logger.

cowbird.utils.print_log(msg: str, logger: logging.Logger | None = None, level: int = logging.INFO, **kwargs: Any) → None[source]: Logs the requested message to the logger and optionally enforce printing to the console according to configuration value defined by COWBIRD_LOG_PRINT.

cowbird.utils.raise_log(msg: str, exception: Type[Exception] = Exception, logger: logging.Logger | None = None, level: int = logging.ERROR) → NoReturn[source]

Logs the provided message to the logger and raises the corresponding exception afterwards.

Raises:: exception – whichever exception provided is raised systematically after logging.

cowbird.utils.bool2str(value: Any) → str[source]: Converts value to explicit "true" or "false" str with permissive variants comparison that can represent common falsy or truthy values.

cowbird.utils.islambda(func: Any) → bool[source]: Evaluate if argument is a callable lambda expression.

cowbird.utils.configure_celery(config: pyramid.config.Configurator, config_ini: str) → None[source]

cowbird.utils.get_app_config(container: cowbird.typedefs.AnySettingsContainer) → pyramid.config.Configurator[source]: Generates application configuration with all required utilities and settings configured.

cowbird.utils.get_settings_from_config_ini(config_ini_path: str, section: str | None = None) → cowbird.typedefs.SettingsType[source]: Loads configuration INI settings with additional handling.

cowbird.utils.get_registry(container: cowbird.typedefs.AnyRegistryContainer, nothrow: bool = False) → pyramid.registry.Registry | None[source]: Retrieves the application registry from various containers referencing to it.

cowbird.utils.get_json(response: cowbird.typedefs.AnyResponseType) → cowbird.typedefs.JSON[source]: Retrieves the ‘JSON’ body of a response using the property/callable according to the response’s implementation.

cowbird.utils.get_header(header_name: str, header_container: cowbird.typedefs.AnyHeadersType, default: str | None = None, split: str | List[str] | None = None) → str | None[source]

Retrieves header_name by fuzzy match (independently of upper/lower-case and underscore/dash) from various framework implementations of Headers.

If split is specified, the matched header_name is first split with it and the first item is returned. This allows to parse complex headers (e.g.: text/plain; charset=UTF-8 to text/plain with split=';').

Parameters:

header_name – header to find.
header_container – where to look for header_name.
default – value to returned if header_container is invalid or header_name could not be found.
split – character(s) to use to split the found header_name.

cowbird.utils.convert_response(response: cowbird.typedefs.AnyResponseType) → pyramid.response.Response[source]

Converts a requests.Response object to an equivalent pyramid.response.Response object.

Content of the response is expected to be JSON.

Parameters:: response – response to be converted
Returns:: converted response

cowbird.utils.get_settings(container: cowbird.typedefs.AnySettingsContainer | None, app: bool = False) → cowbird.typedefs.SettingsType[source]

Retrieve application settings from a supported container.

Parameters:

container – supported container with a handle to application settings.
app – allow retrieving from current thread registry if no container was defined.

Returns:

found application settings dictionary.

Raises:

TypeError – when no application settings could be found or unsupported container.

cowbird.utils.fully_qualified_name(obj: Any | Type[Any]) → str[source]: Obtains the '<module>.<name>' full path definition of the object to allow finding and importing it.

cowbird.utils.log_request_format(request: pyramid.request.Request) → str[source]

cowbird.utils.log_request(event: pyramid.events.NewRequest) → None[source]: Subscriber event that logs basic details about the incoming requests.

cowbird.utils.log_exception_tween(handler: Callable[[pyramid.request.Request], cowbird.typedefs.AnyResponseType], registry: pyramid.registry.Registry) → Callable[[pyramid.request.Request], cowbird.typedefs.AnyResponseType][source]

Tween factory that logs any exception before re-raising it.

Application errors are marked as ERROR while non-critical HTTP errors are marked as WARNING.

cowbird.utils.is_json_body(body: Any) → bool[source]

class cowbird.utils.ExtendedEnum[source]

Bases: enum.Enum

Utility enum.Enum methods.

Create an extended enum with these utilities as follows:

class CustomEnum(ExtendedEnum):
    ItemA = "A"
    ItemB = "B"

classmethod names() → List[str][source]: Returns the member names assigned to corresponding enum elements.

classmethod values() → List[cowbird.typedefs.AnyKey][source]: Returns the literal values assigned to corresponding enum elements.

classmethod get(key_or_value: cowbird.typedefs.AnyKey | EnumClassType, default: Any | None = None) → EnumClassType | None[source]

Finds an enum entry by defined name or its value.

Returns the entry directly if it is already a valid enum.

cowbird.utils.SingletonMetaType: typing_extensions.TypeAlias = 'SingletonMeta'[source]

class cowbird.utils.SingletonMeta[source]

Bases: type

A metaclass that creates a Singleton base class when called.

Create a class such that.

class A(object, metaclass=SingletonMeta):
    pass

class B(object, metaclass=SingletonMeta):
    pass

a1 = A()
a2 = A()
b1 = B()
b2 = B()
a1 is a2    # True
b1 is b2    # True
a1 is b1    # False

_instances: Dict[SingletonMetaType, SingletonMetaType][source]

__call__(*args: Any, **kwargs: Any) → SingletonMetaType[source]: Call self as a function.

class cowbird.utils.NullType[source]

Bases: object

Represents a null value to differentiate from None.

__repr__() → str[source]: Return repr(self).

static __nonzero__() → bool[source]

__bool__[source]

__len__[source]

cowbird.utils.null[source]

cowbird.utils.is_null(item: Any) → bool[source]

cowbird.utils.get_config_path(container: cowbird.typedefs.AnySettingsContainer | None = None) → str[source]

cowbird.utils.get_ssl_verify(container: cowbird.typedefs.AnySettingsContainer | None = None) → bool[source]

cowbird.utils.get_timeout(container: cowbird.typedefs.AnySettingsContainer | None = None) → int[source]

cowbird.utils.apply_new_path_permissions(path: str, is_readable: bool, is_writable: bool, is_executable: bool) → None[source]: Applies new permissions to a path, if required.

cowbird.utils.update_filesystem_permissions(permission: int, is_readable: bool, is_writable: bool, is_executable: bool) → int[source]

Applies/remove read, write and execute permissions (on others only) to the input file system permissions.

Note that others permissions are used instead of the user/group permissions, to manage the user’s data access. See Components - Usage of ‘others’ permissions for more details on the usage of others permissions.

cowbird.utils.apply_default_path_ownership(path: str) → None[source]: Applies default ownership to a path, if required.