Remnants/AdministrationItDepartmens
1
0
Fork 0
mirror of https://gitlab.com/MoonTestUse1/AdministrationItDepartmens.git synced 2025-08-14 00:25:46 +02:00
Code Activity

Проверка 09.02.2025

Browse Source
This commit is contained in:
MoonTestUse1
2025-02-09 01:11:49 +06:00
parent ce52f8a23a
commit 0aa3ef8fc2
5827 changed files with 14316 additions and 1906434 deletions
Download Patch File Download Diff File Expand all files Collapse all files

106
venv/Lib/site-packages/uvicorn/config.py
View File

@@ -9,8 +9,10 @@ import os
import socket
import ssl
import sys
from collections.abc import Awaitable
from configparser import RawConfigParser
from pathlib import Path
from typing import Any, Awaitable, Callable, Literal
from typing import IO, Any, Callable, Literal
import click
@@ -123,13 +125,11 @@ def is_dir(path: Path) -> bool:
if not path.is_absolute():
path = path.resolve()
return path.is_dir()
except OSError:
except OSError: # pragma: full coverage
return False
def resolve_reload_patterns(
patterns_list: list[str], directories_list: list[str]
) -> tuple[list[str], list[Path]]:
def resolve_reload_patterns(patterns_list: list[str], directories_list: list[str]) -> tuple[list[str], list[Path]]:
directories: list[Path] = list(set(map(Path, directories_list.copy())))
patterns: list[str] = patterns_list.copy()
@@ -138,7 +138,7 @@ def resolve_reload_patterns(
# Special case for the .* pattern, otherwise this would only match
# hidden directories which is probably undesired
if pattern == ".*":
continue
continue # pragma: py-darwin
patterns.append(pattern)
if is_dir(Path(pattern)):
directories.append(Path(pattern))
@@ -150,15 +150,13 @@ def resolve_reload_patterns(
directories = list(set(directories))
directories = list(map(Path, directories))
directories = list(map(lambda x: x.resolve(), directories))
directories = list(
{reload_path for reload_path in directories if is_dir(reload_path)}
)
directories = list({reload_path for reload_path in directories if is_dir(reload_path)})
children = []
for j in range(len(directories)):
for k in range(j + 1, len(directories)):
for k in range(j + 1, len(directories)): # pragma: full coverage
if directories[j] in directories[k].parents:
children.append(directories[k]) # pragma: py-darwin
children.append(directories[k])
elif directories[k] in directories[j].parents:
children.append(directories[j])
@@ -193,7 +191,7 @@ class Config:
ws_per_message_deflate: bool = True,
lifespan: LifespanType = "auto",
env_file: str | os.PathLike[str] | None = None,
log_config: dict[str, Any] | str | None = LOGGING_CONFIG,
log_config: dict[str, Any] | str | RawConfigParser | IO[Any] | None = LOGGING_CONFIG,
log_level: str | int | None = None,
access_log: bool = True,
use_colors: bool | None = None,
@@ -216,7 +214,7 @@ class Config:
timeout_notify: int = 30,
timeout_graceful_shutdown: int | None = None,
callback_notify: Callable[..., Awaitable[None]] | None = None,
ssl_keyfile: str | None = None,
ssl_keyfile: str | os.PathLike[str] | None = None,
ssl_certfile: str | os.PathLike[str] | None = None,
ssl_keyfile_password: str | None = None,
ssl_version: int = SSL_PROTOCOL_VERSION,
@@ -280,12 +278,9 @@ class Config:
self.reload_includes: list[str] = []
self.reload_excludes: list[str] = []
if (
reload_dirs or reload_includes or reload_excludes
) and not self.should_reload:
if (reload_dirs or reload_includes or reload_excludes) and not self.should_reload:
logger.warning(
"Current configuration will not reload as not all conditions are met, "
"please refer to documentation."
"Current configuration will not reload as not all conditions are met, " "please refer to documentation."
)
if self.should_reload:
@@ -293,30 +288,23 @@ class Config:
reload_includes = _normalize_dirs(reload_includes)
reload_excludes = _normalize_dirs(reload_excludes)
self.reload_includes, self.reload_dirs = resolve_reload_patterns(
reload_includes, reload_dirs
)
self.reload_includes, self.reload_dirs = resolve_reload_patterns(reload_includes, reload_dirs)
self.reload_excludes, self.reload_dirs_excludes = resolve_reload_patterns(
reload_excludes, []
)
self.reload_excludes, self.reload_dirs_excludes = resolve_reload_patterns(reload_excludes, [])
reload_dirs_tmp = self.reload_dirs.copy()
for directory in self.reload_dirs_excludes:
for reload_directory in reload_dirs_tmp:
if (
directory == reload_directory
or directory in reload_directory.parents
):
if directory == reload_directory or directory in reload_directory.parents:
try:
self.reload_dirs.remove(reload_directory)
except ValueError:
except ValueError: # pragma: full coverage
pass
for pattern in self.reload_excludes:
if pattern in self.reload_includes:
self.reload_includes.remove(pattern)
self.reload_includes.remove(pattern) # pragma: full coverage
if not self.reload_dirs:
if reload_dirs:
@@ -343,11 +331,9 @@ class Config:
self.forwarded_allow_ips: list[str] | str
if forwarded_allow_ips is None:
self.forwarded_allow_ips = os.environ.get(
"FORWARDED_ALLOW_IPS", "127.0.0.1"
)
self.forwarded_allow_ips = os.environ.get("FORWARDED_ALLOW_IPS", "127.0.0.1")
else:
self.forwarded_allow_ips = forwarded_allow_ips
self.forwarded_allow_ips = forwarded_allow_ips # pragma: full coverage
if self.reload and self.workers > 1:
logger.warning('"workers" flag is ignored when reloading is enabled.')
@@ -375,18 +361,14 @@ class Config:
if self.log_config is not None:
if isinstance(self.log_config, dict):
if self.use_colors in (True, False):
self.log_config["formatters"]["default"][
"use_colors"
] = self.use_colors
self.log_config["formatters"]["access"][
"use_colors"
] = self.use_colors
self.log_config["formatters"]["default"]["use_colors"] = self.use_colors
self.log_config["formatters"]["access"]["use_colors"] = self.use_colors
logging.config.dictConfig(self.log_config)
elif self.log_config.endswith(".json"):
elif isinstance(self.log_config, str) and self.log_config.endswith(".json"):
with open(self.log_config) as file:
loaded_config = json.load(file)
logging.config.dictConfig(loaded_config)
elif self.log_config.endswith((".yaml", ".yml")):
elif isinstance(self.log_config, str) and self.log_config.endswith((".yaml", ".yml")):
# Install the PyYAML package or the uvicorn[standard] optional
# dependencies to enable this functionality.
import yaml
@@ -397,9 +379,7 @@ class Config:
else:
# See the note about fileConfig() here:
# https://docs.python.org/3/library/logging.config.html#configuration-file-format
logging.config.fileConfig(
self.log_config, disable_existing_loggers=False
)
logging.config.fileConfig(self.log_config, disable_existing_loggers=False)
if self.log_level is not None:
if isinstance(self.log_level, str):
@@ -430,10 +410,7 @@ class Config:
else:
self.ssl = None
encoded_headers = [
(key.lower().encode("latin1"), value.encode("latin1"))
for key, value in self.headers
]
encoded_headers = [(key.lower().encode("latin1"), value.encode("latin1")) for key, value in self.headers]
self.encoded_headers = (
[(b"server", b"uvicorn")] + encoded_headers
if b"server" not in dict(encoded_headers) and self.server_header
@@ -469,8 +446,7 @@ class Config:
else:
if not self.factory:
logger.warning(
"ASGI app factory detected. Using it, "
"but please consider setting the --factory flag explicitly."
"ASGI app factory detected. Using it, " "but please consider setting the --factory flag explicitly."
)
if self.interface == "auto":
@@ -492,9 +468,7 @@ class Config:
if logger.getEffectiveLevel() <= TRACE_LOG_LEVEL:
self.loaded_app = MessageLoggerMiddleware(self.loaded_app)
if self.proxy_headers:
self.loaded_app = ProxyHeadersMiddleware(
self.loaded_app, trusted_hosts=self.forwarded_allow_ips
)
self.loaded_app = ProxyHeadersMiddleware(self.loaded_app, trusted_hosts=self.forwarded_allow_ips)
self.loaded = True
@@ -512,33 +486,25 @@ class Config:
sock.bind(path)
uds_perms = 0o666
os.chmod(self.uds, uds_perms)
except OSError as exc:
except OSError as exc: # pragma: full coverage
logger.error(exc)
sys.exit(1)
message = "Uvicorn running on unix socket %s (Press CTRL+C to quit)"
sock_name_format = "%s"
color_message = (
"Uvicorn running on "
+ click.style(sock_name_format, bold=True)
+ " (Press CTRL+C to quit)"
)
color_message = "Uvicorn running on " + click.style(sock_name_format, bold=True) + " (Press CTRL+C to quit)"
logger_args = [self.uds]
elif self.fd: # pragma: py-win32
sock = socket.fromfd(self.fd, socket.AF_UNIX, socket.SOCK_STREAM)
message = "Uvicorn running on socket %s (Press CTRL+C to quit)"
fd_name_format = "%s"
color_message = (
"Uvicorn running on "
+ click.style(fd_name_format, bold=True)
+ " (Press CTRL+C to quit)"
)
color_message = "Uvicorn running on " + click.style(fd_name_format, bold=True) + " (Press CTRL+C to quit)"
logger_args = [sock.getsockname()]
else:
family = socket.AF_INET
addr_format = "%s://%s:%d"
if self.host and ":" in self.host: # pragma: py-win32
if self.host and ":" in self.host: # pragma: full coverage
# It's an IPv6 address.
family = socket.AF_INET6
addr_format = "%s://[%s]:%d"
@@ -547,16 +513,12 @@ class Config:
sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
try:
sock.bind((self.host, self.port))
except OSError as exc:
except OSError as exc: # pragma: full coverage
logger.error(exc)
sys.exit(1)
message = f"Uvicorn running on {addr_format} (Press CTRL+C to quit)"
color_message = (
"Uvicorn running on "
+ click.style(addr_format, bold=True)
+ " (Press CTRL+C to quit)"
)
color_message = "Uvicorn running on " + click.style(addr_format, bold=True) + " (Press CTRL+C to quit)"
protocol_name = "https" if self.is_ssl else "http"
logger_args = [protocol_name, self.host, sock.getsockname()[1]]
logger.info(message, *logger_args, extra={"color_message": color_message})