Add QueryParams type

mreg_cli/api/abstracts.py

@@ -19,7 +19,7 @@
     PatchError,
 )
 from mreg_cli.outputmanager import OutputManager
-from mreg_cli.types import JsonMapping
+from mreg_cli.types import JsonMapping, QueryParams
 from mreg_cli.utilities.api import (
     delete,
     get,
@@ -308,15 +308,15 @@ def get_list_by_field(
 
         :returns: A list of objects if found, an empty list otherwise.
         """
-        params = {field: value}
+        params: QueryParams = {field: value}
         if ordering:
             params["ordering"] = ordering
 
         return get_typed(cls.endpoint(), list[cls], params=params, limit=limit)
 
     @classmethod
     def get_by_query(
-        cls, query: dict[str, str], ordering: str | None = None, limit: int | None = 500
+        cls, query: QueryParams, ordering: str | None = None, limit: int | None = 500
     ) -> list[Self]:
         """Get a list of objects by a query.
 
@@ -329,12 +329,12 @@ def get_by_query(
         if ordering:
             query["ordering"] = ordering
 
-        return get_typed(cls.endpoint().with_query(query), list[cls], limit=limit)
+        return get_typed(cls.endpoint(), list[cls], query, limit=limit)
 
     @classmethod
     def get_by_query_unique_or_raise(
         cls,
-        query: dict[str, str],
+        query: QueryParams,
         exc_type: type[Exception] = EntityNotFound,
         exc_message: str | None = None,
     ) -> Self:
@@ -358,7 +358,7 @@ def get_by_query_unique_or_raise(
     @classmethod
     def get_by_query_unique_and_raise(
         cls,
-        query: dict[str, str],
+        query: QueryParams,
         exc_type: type[Exception] = EntityAlreadyExists,
         exc_message: str | None = None,
     ) -> None:
@@ -380,7 +380,7 @@ def get_by_query_unique_and_raise(
         return None
 
     @classmethod
-    def get_by_query_unique(cls, data: dict[str, str]) -> Self | None:
+    def get_by_query_unique(cls, data: QueryParams) -> Self | None:
         """Get an object with the given data.
 
         :param data: The data to search for.

mreg_cli/api/endpoints.py

@@ -126,12 +126,3 @@ def with_params(self, *params: str | int) -> str:
             )
         encoded_params = (quote(str(param)) for param in params)
         return self.value.format(*encoded_params)
-
-    def with_query(self, query: dict[str, str]) -> str:
-        """Construct and return an endpoint URL with a query string.
-
-        :param query: A dictionary of query parameters.
-        :returns: A fully constructed endpoint URL with a query string.
-        """
-        query_string = "&".join(f"{quote(key)}={quote(value)}" for key, value in query.items())
-        return f"{self.value}?{query_string}"

mreg_cli/api/history.py

@@ -12,6 +12,7 @@
 from mreg_cli.api.endpoints import Endpoint
 from mreg_cli.exceptions import EntityNotFound, InternalError
 from mreg_cli.outputmanager import OutputManager
+from mreg_cli.types import QueryParams
 from mreg_cli.utilities.api import get_typed
 
 
@@ -125,7 +126,7 @@ def output_multiple(cls, basename: str, items: list[HistoryItem]) -> None:
     @classmethod
     def get(cls, name: str, resource: HistoryResource) -> list[Self]:
         """Get history items for a resource."""
-        params: dict[str, str] = {"resource": resource.resource(), "name": name}
+        params: QueryParams = {"resource": resource.resource(), "name": name}
         ret = get_typed(Endpoint.History, list[cls], params=params)
         if len(ret) == 0:
             raise EntityNotFound(f"No history found for {name}")

mreg_cli/api/models.py

@@ -42,7 +42,7 @@
     ValidationError,
 )
 from mreg_cli.outputmanager import OutputManager
-from mreg_cli.types import IP_AddressT, IP_NetworkT, IP_Version
+from mreg_cli.types import IP_AddressT, IP_NetworkT, IP_Version, QueryParams
 from mreg_cli.utilities.api import (
     delete,
     get,
@@ -717,7 +717,7 @@ def update_soa(
         :param expire: The expire interval for the zone.
         :param soa_ttl: The TTL for the zone.
         """
-        params: dict[str, str | int | None] = {
+        params: QueryParams = {
             "primary_ns": primary_ns,
             "email": email,
             "serialno": serialno,
@@ -2691,7 +2691,7 @@ def add_ip(self, ip: IP_AddressT, mac: MACAddressField | None = None) -> Host:
 
         :returns: A new Host object fetched from the API with the updated IP address.
         """
-        params: dict[str, str | None] = {"ipaddress": str(ip), "host": str(self.id)}
+        params: QueryParams = {"ipaddress": str(ip), "host": str(self.id)}
         if mac:
             params["macaddress"] = mac.address
 
@@ -3109,7 +3109,7 @@ def endpoint(cls) -> Endpoint:
         return Endpoint.Hosts
 
     @classmethod
-    def get(cls, params: dict[str, Any] | None = None) -> HostList:
+    def get(cls, params: QueryParams | None = None) -> HostList:
         """Get a list of hosts.
 
         :param params: Optional parameters to pass to the API.

mreg_cli/commands/host_submodules/core.py

@@ -36,7 +36,7 @@
     PatchError,
 )
 from mreg_cli.outputmanager import OutputManager
-from mreg_cli.types import Flag
+from mreg_cli.types import Flag, JsonMapping, QueryParams
 from mreg_cli.utilities.shared import convert_wildcard_to_regex
 
 
@@ -118,7 +118,7 @@ def add(args: argparse.Namespace) -> None:
         except ValueError as e:
             raise InputFailure(f"invalid MAC address: {macaddress}") from e
 
-    data: dict[str, str | None] = {
+    data: JsonMapping = {
         "name": hname.hostname,
         "contact": args.contact or None,
         "comment": args.comment or None,
@@ -377,7 +377,7 @@ def _add_param(param: str, value: str) -> None:
     if not any([args.name, args.comment, args.contact]):
         raise InputFailure("Need at least one search critera")
 
-    params: dict[str, str | int] = {
+    params: QueryParams = {
         "ordering": "name",
     }
 

mreg_cli/commands/host_submodules/rr.py

@@ -67,7 +67,7 @@
     PatchError,
 )
 from mreg_cli.outputmanager import OutputManager
-from mreg_cli.types import Flag
+from mreg_cli.types import Flag, QueryParams
 
 
 @command_registry.register_command(
@@ -340,7 +340,7 @@ def naptr_add(args: argparse.Namespace) -> None:
     :param args: argparse.Namespace (name, preference, order, flag, service, regex, replacement)
     """
     host = Host.get_by_any_means_or_raise(args.name)
-    arg_data: dict[str, str | int] = {
+    params: QueryParams = {
         "preference": args.preference,
         "order": args.order,
         "flag": args.flag,
@@ -349,15 +349,10 @@ def naptr_add(args: argparse.Namespace) -> None:
         "replacement": args.replacement,
         "host": host.id,
     }
-
-    # Query parameters must be strings...
-    search_data: dict[str, str] = {k: str(v) for k, v in arg_data.items()}
-
-    existing_naptr = NAPTR.get_by_query_unique(search_data)
+    existing_naptr = NAPTR.get_by_query_unique(params)
     if existing_naptr:
         raise EntityAlreadyExists(f"{host} already has that NAPTR defined.")
-
-    NAPTR.create(params=arg_data)
+    NAPTR.create(params=params)
     OutputManager().add_ok(f"Added NAPTR record to {host.name.hostname}.")
 
 
@@ -622,7 +617,7 @@ def srv_add(args: argparse.Namespace) -> None:
     if not hzone:
         raise EntityNotFound(f"{host} isn't in a zone controlled by MREG")
 
-    data: dict[str, str] = {
+    data: QueryParams = {
         "name": sname.hostname,
         "priority": args.priority,
         "weight": args.weight,
@@ -634,10 +629,7 @@ def srv_add(args: argparse.Namespace) -> None:
     existing_srv = Srv.get_by_query_unique(data)
     if existing_srv:
         raise EntityAlreadyExists(f"{sname} already has that SRV defined.")
-
-    arg_data: dict[str, str | None] = {k: v for k, v in data.items()}
-
-    Srv.create(arg_data)
+    Srv.create(data)
     OutputManager().add_ok(f"Added SRV record {sname} with target {host}.")
 
 
@@ -679,7 +671,7 @@ def srv_remove(args: argparse.Namespace) -> None:
     host = Host.get_by_any_means_or_raise(args.host)
     sname = HostT(hostname=args.name)
 
-    data: dict[str, str] = {
+    data: QueryParams = {
         "name": sname.hostname,
         "host": str(host.id),
         "priority": args.priority,
@@ -739,18 +731,18 @@ def sshfp_add(args: argparse.Namespace) -> None:
     """
     host = Host.get_by_any_means_or_raise(args.name)
 
-    data: dict[str, str] = {
+    data: QueryParams = {
         "algorithm": args.algorithm,
         "hash_type": args.hash_type,
         "fingerprint": args.fingerprint,
-        "host": str(host.id),
+        "host": host.id,
     }
 
     existing_sshfp = SSHFP.get_by_query_unique(data)
     if existing_sshfp:
         raise EntityAlreadyExists(f"{host} already has that SSHFP defined.")
 
-    arg_data: dict[str, str | None] = {k: v for k, v in data.items()}
+    arg_data = {k: v for k, v in data.items()}
     SSHFP.create(arg_data)
     OutputManager().add_ok(f"Added SSHFP record for {host.name.hostname}.")
 

mreg_cli/commands/network.py

@@ -17,7 +17,7 @@
     NetworkOverlap,
 )
 from mreg_cli.outputmanager import OutputManager
-from mreg_cli.types import Flag
+from mreg_cli.types import Flag, QueryParams
 from mreg_cli.utilities.shared import convert_wildcard_to_regex, string_to_int
 from mreg_cli.utilities.validators import is_valid_category_tag, is_valid_location_tag
 
@@ -191,7 +191,7 @@ def find(args: argparse.Namespace) -> None:
         addr = IPAddressField(address=ip_arg)
         networks = [Network.get_by_ip_or_raise(addr.address)]
     else:
-        params: dict[str, str] = {}
+        params: QueryParams = {}
         param_names = [
             "network",
             "description",

mreg_cli/commands/permission.py

@@ -10,7 +10,7 @@
 from mreg_cli.commands.registry import CommandRegistry
 from mreg_cli.exceptions import DeleteError, EntityNotFound
 from mreg_cli.outputmanager import OutputManager
-from mreg_cli.types import Flag
+from mreg_cli.types import Flag, QueryParams
 from mreg_cli.utilities.shared import convert_wildcard_to_regex
 
 command_registry = CommandRegistry()
@@ -46,7 +46,7 @@ def network_list(args: argparse.Namespace) -> None:
     """
     permission_list: list[Permission] = []
 
-    params: dict[str, str] = {}
+    params: QueryParams = {}
     if args.group is not None:
         param, value = convert_wildcard_to_regex("group", args.group)
         params[param] = value

mreg_cli/types.py

@@ -11,6 +11,7 @@
     Any,
     Literal,
     Mapping,
+    MutableMapping,
     NamedTuple,
     Sequence,
     TypeAlias,
@@ -90,6 +91,7 @@ def json_custom_error_validator(
     ],
 )
 JsonMapping = Mapping[str, Json]
+QueryParams = MutableMapping[str, str | int | None]
 
 
 class Flag: