Skip to content
Snippets Groups Projects
Verified Commit 6e82c388 authored by Karel van Klink's avatar Karel van Klink :smiley_cat:
Browse files

add typing to services

parent 6bba6d1d
No related branches found
No related tags found
1 merge request!36Add linter tools, and resolve all linting errors
import ipaddress import ipaddress
from typing import Union from typing import Union, Optional
from pydantic import BaseSettings from pydantic import BaseSettings
...@@ -25,7 +25,7 @@ class HostAddresses(BaseSettings): ...@@ -25,7 +25,7 @@ class HostAddresses(BaseSettings):
v6: ipaddress.IPv6Address v6: ipaddress.IPv6Address
def new_service_networks(service_type: str = "", comment: str = "", extattrs=None) -> ServiceNetworks: def new_service_networks(service_type: str = "", comment: str = "", extattrs: dict = None) -> ServiceNetworks:
if extattrs is None: if extattrs is None:
extattrs = {} extattrs = {}
v4_service_network = _ipam.allocate_service_ipv4_network( v4_service_network = _ipam.allocate_service_ipv4_network(
...@@ -40,10 +40,10 @@ def new_service_networks(service_type: str = "", comment: str = "", extattrs=Non ...@@ -40,10 +40,10 @@ def new_service_networks(service_type: str = "", comment: str = "", extattrs=Non
def new_service_host( def new_service_host(
hostname: str, hostname: str,
service_type: str = "", service_type: str = "",
service_networks: ServiceNetworks = None, service_networks: Optional[ServiceNetworks] = None,
host_addresses: HostAddresses = None, host_addresses: Optional[HostAddresses] = None,
cname_aliases=None, cname_aliases: list = None,
extattrs=None, extattrs: dict = None,
) -> HostAddresses: ) -> HostAddresses:
if extattrs is None: if extattrs is None:
extattrs = {} extattrs = {}
......
...@@ -3,14 +3,14 @@ import requests ...@@ -3,14 +3,14 @@ import requests
from gso import settings from gso import settings
def import_new_router(router_name, oss_params=settings.OSSParams) -> None: def import_new_router(router_name: str, oss_params=settings.OSSParams) -> None:
r = requests.post( r = requests.post(
f"{oss_params.RESOURCE_MANAGER_API_PREFIX}" f"/api/interfaces/initialize-router/{router_name}", timeout=10000 f"{oss_params.RESOURCE_MANAGER_API_PREFIX}" f"/api/interfaces/initialize-router/{router_name}", timeout=10000
) )
r.raise_for_status() r.raise_for_status()
def next_lag(router_name, oss_params=settings.OSSParams) -> dict: def next_lag(router_name: str, oss_params=settings.OSSParams) -> dict:
r = requests.post( r = requests.post(
f"{oss_params.RESOURCE_MANAGER_API_PREFIX}" f"/api/interfaces/next-lag/{router_name}", timeout=10000 f"{oss_params.RESOURCE_MANAGER_API_PREFIX}" f"/api/interfaces/next-lag/{router_name}", timeout=10000
) )
...@@ -19,7 +19,7 @@ def next_lag(router_name, oss_params=settings.OSSParams) -> dict: ...@@ -19,7 +19,7 @@ def next_lag(router_name, oss_params=settings.OSSParams) -> dict:
return response["name"] return response["name"]
def next_physical(router_name, lag_name, oss_params=settings.OSSParams) -> dict: def next_physical(router_name: str, lag_name: str, oss_params=settings.OSSParams) -> dict:
# TODO: speed needed (if first interface) # TODO: speed needed (if first interface)
r = requests.post( r = requests.post(
f"{oss_params.RESOURCE_MANAGER_API_PREFIX}" f"/api/interfaces/next-physical/{router_name}/{lag_name}", f"{oss_params.RESOURCE_MANAGER_API_PREFIX}" f"/api/interfaces/next-physical/{router_name}/{lag_name}",
......
...@@ -25,6 +25,7 @@ exclude = ''' ...@@ -25,6 +25,7 @@ exclude = '''
[tool.mypy] [tool.mypy]
exclude = [ exclude = [
"venv", "venv",
"test/*",
"gso/services/_ipam.py" # TODO: remove "gso/services/_ipam.py" # TODO: remove
] ]
ignore_missing_imports = true ignore_missing_imports = true
......
0% Loading or .
You are about to add 0 people to the discussion. Proceed with caution.
Please register or to comment