Update stubs

This commit is contained in:
Trinh Anh Ngoc
2022-11-05 12:46:29 +07:00
parent 95ca10a411
commit c86a0fa0db
6 changed files with 168 additions and 148 deletions

View File

@@ -1,11 +1,11 @@
from typing import Any, Optional
from logging import Logger
from typing import Any, Literal
_logger: Any
RPC_VERSION_1: Any
RPC_VERSION_1: dict[str, Any]
def exp_login(db: Any, login: Any, password: Any): ...
def exp_authenticate(db: Any, login: Any, password: Any, user_agent_env: Any): ...
def exp_version(): ...
def exp_about(extended: bool = ...): ...
def exp_set_loglevel(loglevel: Any, logger: Optional[Any] = ...): ...
def dispatch(method: Any, params: Any): ...
def exp_login(db: str, login: str, password: str) -> int: ...
def exp_authenticate(db: str, login: str, password: str, user_agent_env: dict | None) -> int: ...
def exp_version() -> dict[str, Any]: ...
def exp_about(extended: bool = ...) -> str | tuple[str, str]: ...
def exp_set_loglevel(loglevel, logger: Logger | None = ...) -> Literal[True]: ...
def dispatch(method: str, params): ...

View File

@@ -1,31 +1,35 @@
from functools import wraps as wraps
from typing import Any, Optional
from typing import Any, Callable, IO, Iterable, Literal, TypeVar
_logger: Any
from ..sql_db import Cursor
_CallableT = TypeVar('_CallableT', bound=Callable)
class DatabaseExists(Warning): ...
def check_db_management_enabled(method: Any): ...
def check_super(passwd: Any): ...
def _initialize_db(id: Any, db_name: Any, demo: Any, lang: Any, user_password: Any, login: str = ..., country_code: Optional[Any] = ..., phone: Optional[Any] = ...) -> None: ...
def _create_empty_database(name: Any) -> None: ...
def exp_create_database(db_name: Any, demo: Any, lang: Any, user_password: str = ..., login: str = ..., country_code: Optional[Any] = ..., phone: Optional[Any] = ...): ...
def exp_duplicate_database(db_original_name: Any, db_name: Any): ...
def _drop_conn(cr: Any, db_name: Any) -> None: ...
def exp_drop(db_name: Any): ...
def exp_dump(db_name: Any, format: Any): ...
def dump_db_manifest(cr: Any): ...
def dump_db(db_name: Any, stream: Any, backup_format: str = ...): ...
def exp_restore(db_name: Any, data: Any, copy: bool = ...): ...
def restore_db(db: Any, dump_file: Any, copy: bool = ...) -> None: ...
def exp_rename(old_name: Any, new_name: Any): ...
def exp_change_admin_password(new_password: Any): ...
def exp_migrate_databases(databases: Any): ...
def exp_db_exist(db_name: Any): ...
def list_dbs(force: bool = ...): ...
def list_db_incompatible(databases: Any): ...
def exp_list(document: bool = ...): ...
def exp_list_lang(): ...
def exp_list_countries(): ...
def exp_server_version(): ...
def dispatch(method: Any, params: Any): ...
def check_db_management_enabled(method: _CallableT) -> _CallableT: ...
def check_super(passwd: str) -> Literal[True]: ...
def _initialize_db(id, db_name: str, demo: bool, lang: str, user_password: str, login: str = ...,
country_code: str | None = ..., phone: str | None = ...) -> None: ...
def _create_empty_database(name: str) -> None: ...
def exp_create_database(db_name: str, demo: bool, lang: str, user_password: str = ..., login: str = ...,
country_code: str | None = ..., phone: str | None = ...) -> Literal[True]: ...
def exp_duplicate_database(db_original_name: str, db_name: str) -> Literal[True]: ...
def _drop_conn(cr: Cursor, db_name: str) -> None: ...
def exp_drop(db_name: str) -> bool: ...
def exp_dump(db_name: str, format: str) -> str: ...
def dump_db_manifest(cr: Cursor) -> dict[str, Any]: ...
def dump_db(db_name: str, stream, backup_format: str = ...) -> IO | None: ...
def exp_restore(db_name: str, data, copy: bool = ...) -> Literal[True]: ...
def restore_db(db: str, dump_file: str, copy: bool = ...) -> None: ...
def exp_rename(old_name: str, new_name: str) -> Literal[True]: ...
def exp_change_admin_password(new_password: str) -> Literal[True]: ...
def exp_migrate_databases(databases: Iterable[str]) -> Literal[True]: ...
def exp_db_exist(db_name: str) -> bool: ...
def list_dbs(force: bool = ...) -> list[str]: ...
def list_db_incompatible(databases: Iterable[str]) -> list[str]: ...
def exp_list(document: bool = ...) -> list[str]: ...
def exp_list_lang() -> list[tuple[str, str]]: ...
def exp_list_countries() -> list[tuple[str, str]]: ...
def exp_server_version() -> str: ...
def dispatch(method: str, params): ...

View File

@@ -1,11 +1,14 @@
from typing import Any, Optional
from typing import Callable, TypeVar
_logger: Any
PG_CONCURRENCY_ERRORS_TO_RETRY: Any
from ..sql_db import Cursor
PG_CONCURRENCY_ERRORS_TO_RETRY: tuple[str, str, str]
MAX_TRIES_ON_CONCURRENCY_FAILURE: int
def dispatch(method: Any, params: Any): ...
def check(f: Any): ...
def execute_cr(cr: Any, uid: Any, obj: Any, method: Any, *args: Any, **kw: Any): ...
def execute_kw(db: Any, uid: Any, obj: Any, method: Any, args: Any, kw: Optional[Any] = ...): ...
def execute(db: Any, uid: Any, obj: Any, method: Any, *args: Any, **kw: Any): ...
_CallableT = TypeVar('_CallableT', bound=Callable)
def dispatch(method: str, params): ...
def check(f: _CallableT) -> _CallableT: ...
def execute_cr(cr: Cursor, uid: int, obj: str, method: str, *args, **kw): ...
def execute_kw(db: str, uid: int, obj: str, method: str, args, kw: dict | None = ...): ...
def execute(db: str, uid: int, obj: str, method: str, *args, **kw): ...

View File

@@ -1,5 +1,8 @@
from typing import Any
from typing import Literal
def check(db: Any, uid: Any, passwd: Any): ...
def compute_session_token(session: Any, env: Any): ...
def check_session(session: Any, env: Any): ...
from ..api import Environment
from ..http import OpenERPSession
def check(db: str, uid: int, passwd: str) -> None: ...
def compute_session_token(session: OpenERPSession, env: Environment) -> str | Literal[False]: ...
def check_session(session: OpenERPSession, env: Environment) -> bool: ...

View File

@@ -1,147 +1,158 @@
import werkzeug.serving
from ..tests import runner as runner
from socket import socket as socket_
from threading import Semaphore, Thread
from gevent.pywsgi import WSGIServer
from itertools import chain as chain
from typing import Any, Optional
from typing import Any, Callable, Iterable, Literal, TypeVar
import werkzeug.serving
from inotify.adapters import InotifyTrees
from psutil import Process
from watchdog.observers import Observer
from ..modules.registry import Registry
from ..sql_db import Cursor
from ..tests import runner as runner
_WorkerT = TypeVar('_WorkerT', bound=Worker)
INOTIFY_LISTEN_EVENTS: Any
_logger: Any
SLEEP_INTERVAL: int
def memory_info(process: Any): ...
def memory_info(process: Process): ...
def set_limit_memory_hard() -> None: ...
def empty_pipe(fd: Any) -> None: ...
def empty_pipe(fd: int) -> None: ...
class LoggingBaseWSGIServerMixIn:
def handle_error(self, request: Any, client_address: Any) -> None: ...
def handle_error(self, request, client_address) -> None: ...
class BaseWSGIServerNoBind(LoggingBaseWSGIServerMixIn, werkzeug.serving.BaseWSGIServer):
def __init__(self, app: Any) -> None: ...
def __init__(self, app) -> None: ...
def server_activate(self) -> None: ...
class RequestHandler(werkzeug.serving.WSGIRequestHandler):
timeout: int = ...
def setup(self) -> None: ...
class ThreadedWSGIServerReloadable(LoggingBaseWSGIServerMixIn, werkzeug.serving.ThreadedWSGIServer):
max_http_threads: Any = ...
http_threads_sem: Any = ...
daemon_threads: bool = ...
def __init__(self, host: Any, port: Any, app: Any) -> None: ...
reload_socket: bool = ...
socket: Any = ...
max_http_threads: Any
http_threads_sem: Semaphore
daemon_threads: bool
def __init__(self, host: str, port: int, app) -> None: ...
reload_socket: bool
socket: socket_
def server_bind(self) -> None: ...
def server_activate(self) -> None: ...
def process_request(self, request: Any, client_address: Any) -> None: ...
def process_request(self, request, client_address) -> None: ...
def _handle_request_noblock(self) -> None: ...
def shutdown_request(self, request: Any) -> None: ...
def shutdown_request(self, request) -> None: ...
class FSWatcherBase:
def handle_file(self, path: Any): ...
def handle_file(self, path: str) -> Literal[True]: ...
class FSWatcherWatchdog(FSWatcherBase):
observer: Any = ...
observer: Observer
def __init__(self) -> None: ...
def dispatch(self, event: Any) -> None: ...
def dispatch(self, event) -> None: ...
def start(self) -> None: ...
def stop(self) -> None: ...
class FSWatcherInotify(FSWatcherBase):
started: bool = ...
watcher: Any = ...
started: bool
watcher: InotifyTrees
def __init__(self) -> None: ...
def run(self) -> None: ...
thread: Any = ...
thread: Thread
def start(self) -> None: ...
def stop(self) -> None: ...
class CommonServer:
app: Any = ...
interface: Any = ...
port: Any = ...
pid: Any = ...
def __init__(self, app: Any) -> None: ...
def close_socket(self, sock: Any) -> None: ...
app: Any
interface: str
port: int
pid: int
def __init__(self, app) -> None: ...
def close_socket(self, sock: socket_) -> None: ...
class ThreadedServer(CommonServer):
main_thread_id: Any = ...
quit_signals_received: int = ...
httpd: Any = ...
limits_reached_threads: Any = ...
limit_reached_time: Any = ...
def __init__(self, app: Any) -> None: ...
def signal_handler(self, sig: Any, frame: Any) -> None: ...
main_thread_id: int | None
quit_signals_received: int
httpd: ThreadedWSGIServerReloadable | None
limits_reached_threads: set[Thread]
limit_reached_time: float | None
def __init__(self, app) -> None: ...
def signal_handler(self, sig, frame) -> None: ...
def process_limit(self) -> None: ...
def cron_thread(self, number: Any) -> None: ...
def cron_thread(self, number) -> None: ...
def cron_spawn(self) -> None: ...
def http_thread(self): ...
def http_spawn(self) -> None: ...
def start(self, stop: bool = ...): ...
def stop(self) -> None: ...
def run(self, preload: Optional[Any] = ..., stop: bool = ...): ...
def run(self, preload: Any | None = ..., stop: bool = ...): ...
def reload(self) -> None: ...
class GeventServer(CommonServer):
port: Any = ...
httpd: Any = ...
def __init__(self, app: Any) -> None: ...
port: int
httpd: WSGIServer | None
def __init__(self, app) -> None: ...
def process_limits(self) -> None: ...
ppid: Any = ...
ppid: int
def watchdog(self, beat: int = ...) -> None: ...
client_address: Any = ...
client_address: Any
def start(self): ...
def stop(self) -> None: ...
def run(self, preload: Any, stop: Any) -> None: ...
def run(self, preload, stop: bool) -> None: ...
class PreforkServer(CommonServer):
address: Any = ...
population: Any = ...
timeout: Any = ...
limit_request: Any = ...
cron_timeout: Any = ...
beat: int = ...
app: Any = ...
pid: Any = ...
socket: Any = ...
workers_http: Any = ...
workers_cron: Any = ...
workers: Any = ...
generation: int = ...
queue: Any = ...
long_polling_pid: Any = ...
def __init__(self, app: Any) -> None: ...
def pipe_new(self): ...
def pipe_ping(self, pipe: Any) -> None: ...
def signal_handler(self, sig: Any, frame: Any) -> None: ...
def worker_spawn(self, klass: Any, workers_registry: Any): ...
address: Any
population: int
timeout: int
limit_request: int
cron_timeout: int
beat: int
app: Any
pid: int
socket: socket_ | None
workers_http: dict[int, WorkerHTTP]
workers_cron: dict[int, WorkerCron]
workers: dict[int, Worker]
generation: int
queue: list
long_polling_pid: int | None
def __init__(self, app) -> None: ...
def pipe_new(self) -> tuple[int, int]: ...
def pipe_ping(self, pipe: tuple[int, int]) -> None: ...
def signal_handler(self, sig: int, frame) -> None: ...
def worker_spawn(self, klass: Callable[..., _WorkerT], workers_registry: dict[int, _WorkerT]) -> _WorkerT | None: ...
def long_polling_spawn(self) -> None: ...
def worker_pop(self, pid: Any) -> None: ...
def worker_kill(self, pid: Any, sig: Any) -> None: ...
def worker_pop(self, pid: int) -> None: ...
def worker_kill(self, pid: int, sig: int) -> None: ...
def process_signals(self) -> None: ...
def process_zombie(self) -> None: ...
def process_timeout(self) -> None: ...
def process_spawn(self) -> None: ...
def sleep(self) -> None: ...
pipe: Any = ...
pipe: tuple[int, int]
def start(self) -> None: ...
def stop(self, graceful: bool = ...) -> None: ...
def run(self, preload: Any, stop: Any): ...
def run(self, preload, stop: bool): ...
class Worker:
multi: Any = ...
watchdog_time: Any = ...
watchdog_pipe: Any = ...
eintr_pipe: Any = ...
watchdog_timeout: Any = ...
ppid: Any = ...
pid: Any = ...
alive: bool = ...
request_max: Any = ...
request_count: int = ...
def __init__(self, multi: Any) -> None: ...
multi: PreforkServer
watchdog_time: float
watchdog_pipe: tuple[int, int]
eintr_pipe: tuple[int, int]
watchdog_timeout: Any
ppid: int
pid: int | None
alive: bool
request_max: Any
request_count: int
def __init__(self, multi: PreforkServer) -> None: ...
def setproctitle(self, title: str = ...) -> None: ...
def close(self) -> None: ...
def signal_handler(self, sig: Any, frame: Any) -> None: ...
def signal_time_expired_handler(self, n: Any, stack: Any) -> None: ...
def signal_handler(self, sig: int, frame) -> None: ...
def signal_time_expired_handler(self, n, stack) -> None: ...
def sleep(self) -> None: ...
def check_limits(self) -> None: ...
def process_work(self) -> None: ...
@@ -151,27 +162,27 @@ class Worker:
def _runloop(self) -> None: ...
class WorkerHTTP(Worker):
sock_timeout: Any = ...
def __init__(self, multi: Any) -> None: ...
def process_request(self, client: Any, addr: Any) -> None: ...
sock_timeout: float
def __init__(self, multi: PreforkServer) -> None: ...
def process_request(self, client: socket_, addr) -> None: ...
def process_work(self) -> None: ...
server: Any = ...
server: BaseWSGIServerNoBind
def start(self) -> None: ...
class WorkerCron(Worker):
db_index: int = ...
watchdog_timeout: Any = ...
def __init__(self, multi: Any) -> None: ...
db_index: int
watchdog_timeout: int
def __init__(self, multi: PreforkServer) -> None: ...
def sleep(self) -> None: ...
def _db_list(self): ...
def process_work(self) -> None: ...
def start(self) -> None: ...
server: Any
server: CommonServer | None
def load_server_wide_modules() -> None: ...
def _reexec(updated_modules: Optional[Any] = ...) -> None: ...
def load_test_file_py(registry: Any, test_file: Any) -> None: ...
def preload_registries(dbnames: Any): ...
def start(preload: Optional[Any] = ..., stop: bool = ...): ...
def _reexec(updated_modules: Iterable[str] | None = ...) -> None: ...
def load_test_file_py(registry: Registry, test_file: str) -> None: ...
def preload_registries(dbnames: list[str] | None): ...
def start(preload: list[str] | None = ..., stop: bool = ...): ...
def restart() -> None: ...

View File

@@ -1,17 +1,16 @@
from typing import Any
from werkzeug.middleware.proxy_fix import ProxyFix as ProxyFix_
from typing import Callable
_logger: Any
RPC_FAULT_CODE_CLIENT_ERROR: int
RPC_FAULT_CODE_APPLICATION_ERROR: int
RPC_FAULT_CODE_WARNING: int
RPC_FAULT_CODE_ACCESS_DENIED: int
RPC_FAULT_CODE_ACCESS_ERROR: int
def xmlrpc_handle_exception_int(e: Any): ...
def xmlrpc_handle_exception_string(e: Any): ...
def _patch_xmlrpc_marshaller() -> None: ...
def application_unproxied(environ: Any, start_response: Any): ...
def xmlrpc_handle_exception_int(e: Exception) -> str: ...
def xmlrpc_handle_exception_string(e: Exception) -> str: ...
def application_unproxied(environ, start_response): ...
ProxyFix: Any
ProxyFix: Callable[..., ProxyFix_]
def application(environ: Any, start_response: Any): ...
def application(environ, start_response): ...