NgTech-LTD/gns3-server
2
0
Fork 0
mirror of https://github.com/GNS3/gns3-server.git synced 2026-09-30 07:40:12 +03:00
Code Issues Packages Projects Releases Wiki Activity

fix(typing): resolve mypy errors in agent.web_wireshark.docker_client

Browse Source
This commit is contained in:
Mark Paronyan 2026-09-28 05:21:26 +00:00
parent d25cb5cdca
commit c154d1c41e
No known key found for this signature in database
GPG Key ID: 8DC7338A87FE22F8
2 changed files with 12 additions and 6 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

17
gns3server/agent/web_wireshark/docker_client.py
View File

@ -23,9 +23,16 @@ This module handles Docker API operations via aiohttp Unix socket.
import asyncio
import logging
import sys
from typing import Any
import aiohttp
if sys.version_info >= (3, 11):
from asyncio import timeout as async_timeout
else:
from async_timeout import timeout as async_timeout
from gns3server.utils import parse_version
logger = logging.getLogger(__name__)
@ -132,7 +139,7 @@ class DockerHTTPClient:
url = f"http://docker/v{self._api_version}/{endpoint}"
try:
async with asyncio.timeout(self.REQUEST_TIMEOUT):
async with async_timeout(self.REQUEST_TIMEOUT):
async with session.request(method, url, **kwargs) as response:
if response.status >= 300:
error_text = await response.text()
@ -157,9 +164,9 @@ class DockerHTTPClient:
return await self._request(method, endpoint, check_connection=False, **kwargs)
raise
async def create_network(self, name: str, driver: str = "bridge", subnet: str = None):
async def create_network(self, name: str, driver: str = "bridge", subnet: str | None = None):
"""Create Docker network."""
data = {"Name": name, "Driver": driver}
data: dict[str, Any] = {"Name": name, "Driver": driver}
if subnet:
data["IPAM"] = {"Config": [{"Subnet": subnet}]}
await self._request("POST", "networks/create", json=data)
@ -175,7 +182,7 @@ class DockerHTTPClient:
async def create_container(self, name: str, image: str, **kwargs):
"""Create container."""
data = {"Image": image, "name": name, "HostConfig": {}, "NetworkingConfig": {}}
data: dict[str, Any] = {"Image": image, "name": name, "HostConfig": {}, "NetworkingConfig": {}}
# Handle network config
if "network" in kwargs:
@ -248,7 +255,7 @@ class DockerHTTPClient:
url = f"http://docker/v{self._api_version}/containers/{container_name}/top?ps_args=aux"
try:
async with asyncio.timeout(self.REQUEST_TIMEOUT):
async with async_timeout(self.REQUEST_TIMEOUT):
async with session.get(url) as response:
if response.status >= 300:
error_text = await response.text()

1
pyproject.toml
View File

@ -313,7 +313,6 @@ module = [
"gns3server.agent.gns3_copilot.tools_v2.vpcs_tools_netmiko", # 9
"gns3server.agent.gns3_copilot.utils.message_converters", # 2
"gns3server.agent.gns3_copilot.utils.parse_tool_content", # 1
"gns3server.agent.web_wireshark.docker_client", # 5
"gns3server.agent.web_wireshark.manage_wireshark", # 1
"gns3server.agent.web_wireshark.manager", # 9
"gns3server.agent.web_wireshark.stats", # 4