from _typeshed import HasFileno, OptExcInfo, ReadOnlyBuffer
from _typeshed.wsgi import WSGIApplication
from collections.abc import Callable
from types import ModuleType
from typing import Any, Literal, Protocol, final, overload
from typing_extensions import Self, TypeAlias

import uwsgidecorators

_TrueOrNone: TypeAlias = Literal[True] | None

class _RPCCallable(Protocol):
    def __call__(self, *args: bytes) -> bytes | None: ...

# FIXME: Technically we know the exact layout of _AppsDict and _WorkerDict
#        but TypedDict does not support bytes keys, so for now we use type
#        aliases to a more generic dict
_WorkerDict: TypeAlias = dict[bytes, Any]

SPOOL_IGNORE: Literal[0]
SPOOL_OK: Literal[-2]
SPOOL_RETRY: Literal[-1]
applications: dict[str, WSGIApplication | str] | None
buffer_size: int
cores: int
has_threads: int
hostname: bytes
is_a_reload: bool
loop: bytes | None
magic_table: dict[bytes, bytes]
numproc: int
opt: dict[str, bytes | Literal[True] | list[bytes | Literal[True]]]
sockets: list[int]
started_on: int
unbit: _TrueOrNone
version: bytes
version_info: tuple[int, int, int, int, bytes]
spoolers: tuple[bytes, ...]
queue_size: int

decorators = uwsgidecorators
spooler = uwsgidecorators.manage_spool_request
post_fork_hook = uwsgidecorators.postfork_chain_hook

@final
class SymbolsImporter:
    def find_module(self, fullname: str, /) -> Self | None: ...
    def load_module(self, fullname: str, /) -> ModuleType | None: ...

@final
class SymbolsZipImporter:
    def __init__(self, name: str, /) -> None: ...
    def find_module(self, fullname: str, /) -> Self | None: ...
    def load_module(self, fullname: str, /) -> ModuleType | None: ...

@final
class ZipImporter:
    def __init__(self, name: str, /) -> None: ...
    def find_module(self, fullname: str, /) -> Self | None: ...
    def load_module(self, fullname: str, /) -> ModuleType | None: ...

def accepting(accepting: bool = True, /) -> None: ...
def add_cron(signum: int, minute: int, hour: int, day: int, month: int, weekday: int, /) -> Literal[True]: ...
def add_file_monitor(signum: int, filename: str, /) -> None: ...
def add_rb_timer(signum: int, seconds: int, iterations: int = 0, /) -> None: ...
def add_timer(signum: int, seconds: int, /) -> None: ...
def add_var(key: bytes | str, val: bytes | str, /) -> Literal[True]: ...
def alarm(alarm: str, msg: bytes | ReadOnlyBuffer | str, /) -> None: ...
def async_connect(socket_name: str, /) -> int: ...
def async_sleep(timeout: float, /) -> Literal[b""]: ...
def cache_clear(cache_name: str = ..., /) -> _TrueOrNone: ...
def cache_dec(key: str | bytes, decrement: int = 1, expires: int = 0, cache_name: str = ..., /) -> _TrueOrNone: ...
def cache_del(key: str | bytes, cache_name: str = ..., /) -> _TrueOrNone: ...
def cache_div(key: str | bytes, divisor: int = 2, expires: int = 0, cache_name: str = ..., /) -> _TrueOrNone: ...
def cache_exists(key: str | bytes, cache_name: str = ..., /) -> _TrueOrNone: ...
def cache_get(key: str | bytes, cache_name: str = ..., /) -> bytes | None: ...
def cache_inc(key: str | bytes, increment: int = 1, expires: int = 0, cache_name: str = ..., /) -> _TrueOrNone: ...
def cache_keys(cache_name: str = ..., /) -> list[bytes]: ...
def cache_mul(key: str | bytes, factor: int = 2, expires: int = 0, cache_name: str = ..., /) -> _TrueOrNone: ...
def cache_num(key: str | bytes, cache_name: str = ..., /) -> int: ...
def cache_set(
    key: str | bytes, value: str | bytes | ReadOnlyBuffer, expires: int = 0, cache_name: str = ..., /
) -> _TrueOrNone: ...
def cache_update(
    key: str | bytes, value: str | bytes | ReadOnlyBuffer, expires: int = 0, cache_name: str = ..., /
) -> _TrueOrNone: ...
def queue_get(index: int, /) -> bytes | None: ...
def queue_set(index: int, message: str | bytes | ReadOnlyBuffer, /) -> _TrueOrNone: ...
@overload
def queue_last(num: Literal[0] = 0, /) -> bytes | None: ...  # type: ignore[overload-overlap]
@overload
def queue_last(num: int, /) -> list[bytes | None]: ...
def queue_push(message: str | bytes | ReadOnlyBuffer, /) -> _TrueOrNone: ...
def queue_pull() -> bytes | None: ...
def queue_pop() -> bytes | None: ...
def queue_slot() -> int: ...
def queue_pull_slot() -> int: ...
def snmp_set_community(snmp_community: str, /) -> Literal[True]: ...
def snmp_set_counter32(oid_num: int, value: int, /) -> _TrueOrNone: ...
def snmp_set_counter64(oid_num: int, value: int, /) -> _TrueOrNone: ...
def snmp_set_gauge(oid_num: int, value: int, /) -> _TrueOrNone: ...
def snmp_incr_counter32(oid_num: int, increment: int, /) -> _TrueOrNone: ...
def snmp_incr_counter64(oid_num: int, increment: int, /) -> _TrueOrNone: ...
def snmp_incr_gauge(oid_num: int, increment: int, /) -> _TrueOrNone: ...
def snmp_decr_counter32(oid_num: int, decrement: int, /) -> _TrueOrNone: ...
def snmp_decr_counter64(oid_num: int, decrement: int, /) -> _TrueOrNone: ...
def snmp_decr_gauge(oid_num: int, decrement: int, /) -> _TrueOrNone: ...
@overload
def send_to_spooler(mesage_dict: dict[bytes, bytes], /) -> bytes | None: ...
@overload
def send_to_spooler(
    *, spooler: bytes = ..., priority: bytes = ..., at: bytes = ..., body: bytes = ..., **kwargs: bytes
) -> bytes | None: ...

spool = send_to_spooler

def set_spooler_frequency(frequency: int, /) -> Literal[True]: ...
def spooler_jobs() -> list[bytes]: ...
def spooler_pid() -> int: ...
def spooler_pids() -> list[int]: ...
def spooler_get_task(task_path: str, /) -> dict[bytes, bytes] | None: ...
def call(rpc_name: str, /, *args: bytes) -> bytes | None: ...
def chunked_read(timeout: int = 0, /) -> bytes: ...
def chunked_read_nb() -> bytes: ...
def cl() -> int: ...
def close(fd: int, /) -> None: ...
def connect(socket_name: str, /, timeout: int = 0) -> int: ...
def connection_fd() -> int: ...
def disconnect() -> None: ...
def embedded_data(name: str, /) -> bytes: ...
def extract(name: str, /) -> bytes | None: ...
def farm_get_msg() -> bytes | None: ...
def farm_msg(farm_name: str, message: str | bytes | ReadOnlyBuffer, /) -> None: ...
def get_logvar(key: str | bytes, /) -> bytes | None: ...
def green_schedule() -> Literal[True]: ...
def i_am_the_lord(legion_name: str, /) -> bool: ...
def i_am_the_spooler() -> _TrueOrNone: ...
def in_farm(farm_name: str = ..., /) -> _TrueOrNone: ...
def is_connected(fd: int, /) -> bool: ...
def is_locked(lock_num: int = 0, /) -> bool: ...
def listen_queue(id: int = 0, /) -> int: ...
def lock(lock_num: int = 0, /) -> None: ...
def log(logline: str, /) -> Literal[True]: ...
def log_this_request() -> None: ...
def logsize() -> int: ...
def lord_scroll(legion_name: str, /) -> bytes | None: ...
def masterpid() -> int: ...
def mem() -> tuple[int, int]: ...
def metric_dec(key: str, decrement: int = 1, /) -> _TrueOrNone: ...
def metric_div(key: str, divisor: int = 1, /) -> _TrueOrNone: ...
def metric_get(key: str, /) -> int: ...
def metric_inc(key: str, increment: int = 1, /) -> _TrueOrNone: ...
def metric_mul(key: str, factor: int = 1, /) -> _TrueOrNone: ...
def metric_set(key: str, value: int = 1, /) -> _TrueOrNone: ...
def metric_set_max(key: str, value: int = 1, /) -> _TrueOrNone: ...
def metric_set_min(key: str, value: int = 1, /) -> _TrueOrNone: ...
def micros() -> int: ...
def mule_get_msg(signals: bool = True, farms: bool = True, buffer_size: int = 65536, timeout: int = -1) -> bytes: ...
def mule_id() -> int: ...
@overload
def mule_msg(mesage: str | bytes | ReadOnlyBuffer, /) -> bool: ...
@overload
def mule_msg(mesage: str | bytes | ReadOnlyBuffer, mule_id: int, /) -> bool: ...
@overload
def mule_msg(mesage: str | bytes | ReadOnlyBuffer, farm_name: str, /) -> bool: ...
def offload(filename: str, len: int = 0, /) -> Literal[b""]: ...
def parsefile(filename: str, /) -> dict[bytes, bytes] | None: ...
def ready() -> _TrueOrNone: ...
def ready_fd() -> int: ...
def recv(fd: int, max_size: int = 4096, /) -> bytes | None: ...
@overload
def register_rpc(name: str, func: Callable[[], bytes | None], /) -> Literal[True]: ...
@overload
def register_rpc(name: str, func: Callable[[bytes], bytes | None], /, arg_count: Literal[1]) -> Literal[True]: ...
@overload
def register_rpc(name: str, func: Callable[[bytes, bytes], bytes | None], /, arg_count: Literal[2]) -> Literal[True]: ...
@overload
def register_rpc(name: str, func: _RPCCallable, /, arg_count: int) -> Literal[True]: ...
def register_signal(signum: int, who: str, handler: Callable[[int], Any], /) -> None: ...
def reload() -> _TrueOrNone: ...
def request_id() -> int: ...
def route(router_name: str, router_args: str, /) -> int: ...
def rpc(node: str | bytes, rpc_name: bytes, /, *rpc_args: bytes) -> bytes | None: ...
def rpc_list() -> tuple[bytes, ...]: ...
def scrolls(legion_name: str, /) -> list[bytes] | None: ...
@overload
def send(data: bytes, /) -> _TrueOrNone: ...
@overload
def send(fd: int, data: bytes, /) -> _TrueOrNone: ...
def sendfile(
    filename_or_fd: str | bytes | int | HasFileno, chunk: int = 0, pos: int = 0, /, filesize: int = 0
) -> _TrueOrNone: ...
def set_logvar(key: str | bytes, val: str | bytes, /) -> None: ...
def set_user_harakiri(seconds: int, /) -> None: ...
def set_warning_message(message: str, /) -> Literal[True]: ...
def setprocname(name: str, /) -> None: ...
def signal(signum: int = ..., node: str = ..., /) -> None: ...
def signal_received() -> int: ...
def signal_registered(signum: int, /) -> _TrueOrNone: ...
def signal_wait(signum: int = ..., /) -> Literal[b""]: ...
def start_response(
    status: str, headers: list[tuple[str, str]], exc_info: OptExcInfo | None = ..., /
) -> Callable[[bytes], None]: ...
def stop() -> _TrueOrNone: ...
def suspend() -> Literal[True]: ...
def total_requests() -> int: ...
def unlock(lock_num: int = 0, /) -> None: ...
def wait_fd_read(fd: int, timeout: int = 0, /) -> Literal[b""]: ...
def wait_fd_write(fd: int, timeout: int = 0, /) -> Literal[b""]: ...
def websocket_handshake(key: str | bytes = ..., origin: str | bytes = ..., proto: str | bytes = ..., /) -> None: ...
def websocket_recv() -> bytes: ...
def websocket_recv_nb() -> bytes: ...
def websocket_send(message: str | bytes | ReadOnlyBuffer) -> None: ...
def websocket_send_binary(message: str | bytes | ReadOnlyBuffer) -> None: ...
def worker_id() -> int: ...
def workers() -> tuple[_WorkerDict, ...] | None: ...
def sharedarea_read(id: int, position: int, length: int, /) -> bytes: ...
def sharedarea_write(id: int, position: int, value: str | bytes | ReadOnlyBuffer, /) -> None: ...
def sharedarea_readbyte(id: int, position: int, /) -> int: ...
def sharedarea_writebyte(id: int, position: int, value: int, /) -> None: ...
def sharedarea_read8(id: int, position: int, /) -> int: ...
def sharedarea_write8(id: int, position: int, value: int, /) -> None: ...
def sharedarea_readlong(id: int, position: int, /) -> int: ...
def sharedarea_writelong(id: int, position: int, value: int, /) -> None: ...
def sharedarea_read64(id: int, position: int, /) -> int: ...
def sharedarea_write64(id: int, position: int, value: int, /) -> None: ...
def sharedarea_read32(id: int, position: int, /) -> int: ...
def sharedarea_write32(id: int, position: int, value: int, /) -> None: ...
def sharedarea_read16(id: int, position: int, /) -> int: ...
def sharedarea_write16(id: int, position: int, value: int, /) -> None: ...
def sharedarea_inclong(id: int, position: int, increment: int = 1, /) -> None: ...
def sharedarea_inc64(id: int, position: int, increment: int = 1, /) -> None: ...
def sharedarea_inc32(id: int, position: int, increment: int = 1, /) -> None: ...
def sharedarea_dec64(id: int, position: int, decrement: int = 1, /) -> None: ...
def sharedarea_dec32(id: int, position: int, decrement: int = 1, /) -> None: ...
def sharedarea_rlock(id: int, /) -> None: ...
def sharedarea_wlock(id: int, /) -> None: ...
def sharedarea_unlock(id: int, /) -> None: ...
def sharedarea_object(id: int, /) -> bytearray: ...
def sharedarea_memoryview(id: int, /) -> memoryview: ...