Skip to content
Snippets Groups Projects
Commit 4762a014 authored by Neda Moeini's avatar Neda Moeini
Browse files

Make ruff and mypy happy

parent 34d60e17
No related branches found
No related tags found
1 merge request!1Feature/get graph data from gap
......@@ -5,6 +5,7 @@ from fastapi import FastAPI
from mapping_provider.api.common import router as version_router
from mapping_provider.api.network_graph import router as graph_router
def create_app() -> FastAPI:
"""Create a FastAPI application."""
app = FastAPI(
......@@ -14,9 +15,10 @@ def create_app() -> FastAPI:
# Force configuration to be loaded at startup to avoid issues with missing config
from mapping_provider.dependencies import load_config
_ = load_config()
app.include_router(version_router)
app.include_router(graph_router)
return app
\ No newline at end of file
return app
......@@ -7,5 +7,6 @@ router = APIRouter()
@router.get("/network-graph")
def get_network_graph(config: config_dep):
def get_network_graph(config: config_dep) -> dict[str, list[str] | list[list[str]]]:
"""Network graph endpoint."""
return {"routers": load_routers(config), "trunks": load_trunks(config)}
"""The main module that runs the application."""
import uvicorn
from mapping_provider import create_app
app = create_app()
if __name__ == "__main__":
uvicorn.run("mapping_provider.app:app", host="127.0.0.1", port=8000, reload=True)
\ No newline at end of file
uvicorn.run("mapping_provider.app:app", host="127.0.0.1", port=8000, reload=True)
"""Configuration file for the Mapping Provider."""
import json
from typing import Any, TextIO
import jsonschema
CONFIG_SCHEMA = {
'$schema': 'http://json-schema.org/draft-07/schema#',
'type': 'object',
'properties': {
'aai': {
'type': 'object',
'properties': {
'discovery_endpoint_url': {'type': 'string'},
'client_id': {'type': 'string'},
'secret': {'type': 'string'},
"$schema": "http://json-schema.org/draft-07/schema#",
"type": "object",
"properties": {
"aai": {
"type": "object",
"properties": {
"discovery_endpoint_url": {"type": "string"},
"client_id": {"type": "string"},
"secret": {"type": "string"},
},
'required': ['client_id', 'secret', 'discovery_endpoint_url'],
'additionalProperties': False
"required": ["client_id", "secret", "discovery_endpoint_url"],
"additionalProperties": False,
},
'orchestrator': {
'type': 'object',
'properties': {
'url': {'type': 'string'},
"orchestrator": {
"type": "object",
"properties": {
"url": {"type": "string"},
},
'required': ['url'],
'additionalProperties': False
"required": ["url"],
"additionalProperties": False,
},
},
'required': ['aai', 'orchestrator'],
'additionalProperties': False
"required": ["aai", "orchestrator"],
"additionalProperties": False,
}
def load(f):
def load(f: TextIO) -> dict[str, Any]:
"""loads, validates and returns configuration parameters.
:param f: file-like object that produces the config file
:return:
"""
config = json.loads(f.read())
config: dict[str, Any] = json.loads(f.read())
jsonschema.validate(config, CONFIG_SCHEMA)
return config
......@@ -2,14 +2,15 @@
import os
from functools import lru_cache
from typing import Annotated
from typing import Annotated, Any
from fastapi import Depends
from mapping_provider import config
@lru_cache()
def load_config() -> dict:
@lru_cache
def load_config() -> dict[str, Any]:
"""Load and cache the application configuration."""
config_file = os.environ.get("CONFIG_FILE_NAME", "config.json")
with open(config_file) as f:
......@@ -17,4 +18,4 @@ def load_config() -> dict:
# Dependency for injecting config into routes/services
config_dep = Annotated[dict, Depends(load_config)]
\ No newline at end of file
config_dep = Annotated[dict[str, Any], Depends(load_config)]
"""External services for the mapping provider."""
\ No newline at end of file
"""External services for the mapping provider."""
......@@ -3,6 +3,7 @@ from collections.abc import Callable
from typing import Any
import requests
from requests import Session
from mapping_provider.dependencies import config_dep
......@@ -12,14 +13,15 @@ GRANT_TYPE = "client_credentials"
SCOPE = "openid profile email aarc"
def get_token_endpoint(discovery_endpoint_url: str, session: requests.Session) -> str:
def get_token_endpoint(discovery_endpoint_url: str, session: Session) -> str:
"""Fetch the token endpoint URL from discovery document."""
response = session.get(discovery_endpoint_url)
response.raise_for_status()
return response.json()["token_endpoint"]
data: dict[str, Any] = response.json()
return str(data["token_endpoint"])
def get_token(config: dict, session: requests.Session) -> str:
def get_token(config: dict[str, Any], session: Session) -> str:
"""Retrieve an access token using client credentials flow."""
aai_config = config["aai"]
token_endpoint = get_token_endpoint(aai_config["discovery_endpoint_url"], session)
......@@ -33,17 +35,18 @@ def get_token(config: dict, session: requests.Session) -> str:
},
)
response.raise_for_status()
return response.json()["access_token"]
data: dict[str, Any] = response.json()
return str(data["access_token"])
def make_request(query: str, token: str, config: dict, session: requests.Session) -> dict:
def make_request(query: str, token: str, config: dict[str, Any], session: Session) -> dict[Any, Any]:
"""Make a GraphQL request to the orchestrator API."""
api_url = f"{config['orchestrator']['url']}/api/graphql"
headers = {"Authorization": f"Bearer {token}"}
response = session.post(api_url, headers=headers, json={"query": query})
response.raise_for_status()
data = response.json()
data: dict[str, Any] = response.json()
if "errors" in data:
logger.error(f"GraphQL query returned errors: {data['errors']}. Query: {query}")
raise ValueError(f"GraphQL query errors: {data['errors']}")
......@@ -51,16 +54,16 @@ def make_request(query: str, token: str, config: dict, session: requests.Session
return data
def extract_router(product_block_instances: list[dict]) -> str | None:
def extract_router(product_block_instances: list[dict[str, Any]]) -> str | None:
"""Extract router FQDNs from productBlockInstances."""
for instance in product_block_instances:
for value in instance.get("productBlockInstanceValues", []):
if value.get("field") == "routerFqdn" and value.get("value"):
return value["value"]
return str(value["value"])
return None
def extract_trunk(product_block_instances: list[dict]) -> list[str] | None:
def extract_trunk(product_block_instances: list[dict[str, Any]]) -> list[str] | None:
"""Extract trunks from productBlockInstances."""
fqdns = []
for instance in product_block_instances:
......@@ -75,11 +78,11 @@ def extract_trunk(product_block_instances: list[dict]) -> list[str] | None:
def load_inventory(
config: dict,
token: str,
tag: str,
session: requests.Session,
extractor: Callable[[list[dict]], Any | None],
config: dict[str, Any],
token: str,
tag: str,
session: Session,
extractor: Callable[[list[dict[str, Any]]], Any | None],
) -> list[Any]:
"""
Generic function to load inventory items based on tag and extractor function.
......@@ -143,4 +146,4 @@ def load_trunks(config: config_dep) -> list[list[str]]:
"""Load trunks (edges) from orchestrator."""
with requests.Session() as session:
token = get_token(config, session)
return load_inventory(config, token, tag="IPTRUNK", session=session, extractor=extract_trunk)
\ No newline at end of file
return load_inventory(config, token, tag="IPTRUNK", session=session, extractor=extract_trunk)
[tool.ruff]
line-length = 120
target-version = "py313"
target-version = "py312"
select = ["E", "F", "I", "B", "UP", "N"]
fixable = ["ALL"]
exclude = ["tests", "docs", "build"]
......@@ -9,4 +9,14 @@ exclude = ["tests", "docs", "build"]
python_version = "3.13"
strict = true
warn_unused_ignores = true
warn_return_any = true
\ No newline at end of file
warn_return_any = true
allow_untyped_decorators = true
ignore_missing_imports = true
exclude = [
"venv",
"test/*",
"docs"
]
[[tool.mypy.overrides]]
module = ["requests.*"]
ignore_missing_imports = true
\ No newline at end of file
......@@ -13,7 +13,9 @@ setup(
python_requires=">=3.10",
install_requires=[
"fastapi",
"uvicorn[standard]"
"uvicorn[standard]",
"jsonschema",
"requests",
],
long_description=open("README.md", encoding="utf-8").read(),
long_description_content_type="text/markdown",
......@@ -23,4 +25,4 @@ setup(
"License :: OSI Approved :: MIT License",
"Operating System :: OS Independent",
],
)
\ No newline at end of file
)
[tox]
envlist = lint, typecheck, docs
envlist = py313
[testenv:lint]
description = Lint code with Ruff
deps = ruff
commands = ruff check mapping_provider
[testenv:typecheck]
description = Type-check code with mypy
deps = mypy
commands = mypy mapping_provider
[testenv:docs]
description = Build docs
[testenv]
deps =
sphinx
sphinx-rtd-theme
sphinxcontrib-plantuml
sphinxcontrib-drawio
sphinxcontrib-openapi
commands = sphinx-build -b html docs/source docs/build
-r requirements.txt
commands =
ruff check --respect-gitignore --preview .
ruff format --respect-gitignore --preview --check .
mypy .
0% Loading or .
You are about to add 0 people to the discussion. Proceed with caution.
Please register or to comment