Add module netbox_contact_assignment

changelogs/fragments/contacts.yml

@@ -0,0 +1,2 @@
+minor_changes:
+  - netbox_contact_assignment - New module `#1480 <https://github.com/netbox-community/ansible_modules/pull/1480>`

plugins/module_utils/netbox_tenancy.py

@@ -15,14 +15,74 @@
 NB_TENANTS = "tenants"
 NB_TENANT_GROUPS = "tenant_groups"
 NB_CONTACTS = "contacts"
+NB_CONTACT_ASSIGNMENTS = "contact_assignments"
 NB_CONTACT_GROUPS = "contact_groups"
 NB_CONTACT_ROLES = "contact_roles"
 
 
+OBJECT_ENDPOINTS = {
+    "circuit": "circuits",
+    "cluster": "clusters",
+    "cluster_group": "cluster_groups",
+    "contact": "contacts",
+    "contact_role": "contact_roles",
+    "device": "devices",
+    "location": "locations",
+    "manufacturer": "manufacturers",
+    "power_panel": "power_panels",
+    "provider": "providers",
+    "rack": "racks",
+    "region": "regions",
+    "site": "sites",
+    "site_group": "site_groups",
+    "tenant": "tenants",
+    "virtual_machine": "virtual_machines",
+}
+# See https://netboxlabs.com/docs/netbox/features/contacts/#contacts-1
+OBJECT_TYPES = {
+    "circuit": "circuits.circuit",
+    "cluster": "virtualization.cluster",
+    "cluster_group": "virtualization.clustergroup",
+    "device": "dcim.device",
+    "location": "dcim.location",
+    "manufacturer": "dcim.manufacturer",
+    "power_panel": "dcim.powerpanel",
+    "provider": "circuits.provider",
+    "rack": "dcim.rack",
+    "region": "dcim.region",
+    "site": "dcim.site",
+    "site_group": "dcim.sitegroup",
+    "tenant": "tenancy.tenant",
+    "virtual_machine": "virtualization.virtualmachine",
+}
+OBJECT_NAME_FIELD = {
+    "circuit": "cid",
+    # If unspecified, the default is "name"
+}
+
+
 class NetboxTenancyModule(NetboxModule):
     def __init__(self, module, endpoint):
         super().__init__(module, endpoint)
 
+    def get_object_by_name(self, object_type: str, object_name: str):
+        endpoint = OBJECT_ENDPOINTS[object_type]
+        app = self._find_app(endpoint)
+        nb_app = getattr(self.nb, app)
+        nb_endpoint = getattr(nb_app, endpoint)
+
+        name_field = OBJECT_NAME_FIELD.get(object_type)
+        if name_field is None:
+            name_field = "name"
+
+        query_params = {name_field: object_name}
+        result = self._nb_endpoint_get(nb_endpoint, query_params, object_name)
+        if not result:
+            self._handle_errors(
+                msg="Could not resolve id of %s: %s" % (object_type, object_name)
+            )
+        return result
+
     def run(self):
         """
         This function should have all necessary code for endpoints within the application
@@ -31,7 +91,9 @@ def run(self):
         - tenants
         - tenant groups
         - contacts
+        - contact assignments
         - contact groups
+        - contact roles
         """
         # Used to dynamically set key when returning results
         endpoint_name = ENDPOINT_NAME_MAPPING[self.endpoint]
@@ -45,6 +107,23 @@ def run(self):
 
         data = self.data
 
+        # For ease and consistency of use, the contact assignment module takes the name of the contact, role, and target object rather than an ID or slug.
+        # We must massage the data a bit by looking up the ID corresponding to the given name so that we can pass the ID to the API.
+        if self.endpoint == "contact_assignments":
+            # Not an identifier, just to populate the message field
+            name = f"{data['contact']} -> {data['object_name']}"
+
+            object_type = data["object_type"]
+            obj = self.get_object_by_name(object_type, data["object_name"])
+            contact = self.get_object_by_name("contact", data["contact"])
+            role = self.get_object_by_name("contact_role", data["role"])
+
+            data["object_type"] = OBJECT_TYPES[object_type]
+            data["object_id"] = obj.id
+            del data["object_name"]  # object_id replaces object_name
+            data["contact"] = contact.id
+            data["role"] = role.id
+
         # Used for msg output
         if data.get("name"):
             name = data["name"]
@@ -58,6 +137,12 @@ def run(self):
         object_query_params = self._build_query_params(
             endpoint_name, data, user_query_params
         )
+
+        # For some reason, when creating a new contact assignment, role must be an ID
+        # But when querying contact assignments, the role must be a slug
+        if self.endpoint == "contact_assignments":
+            object_query_params["role"] = role.slug
+
         self.nb_object = self._nb_endpoint_get(nb_endpoint, object_query_params, name)
 
         if self.state == "present":

plugins/module_utils/netbox_utils.py

@@ -114,6 +114,7 @@
         "tenants": {},
         "tenant_groups": {},
         "contacts": {},
+        "contact_assignments": {},
         "contact_groups": {},
         "contact_roles": {},
     },
@@ -366,6 +367,7 @@
     "console_server_ports": "console_server_port",
     "console_server_port_templates": "console_server_port_template",
     "contacts": "contact",
+    "contact_assignments": "contact_assignment",
     "contact_groups": "contact_group",
     "contact_roles": "contact_role",
     "custom_fields": "custom_field",
@@ -476,6 +478,7 @@
     "console_server_port": set(["name", "device"]),
     "console_server_port_template": set(["name", "device_type"]),
     "contact": set(["name", "group"]),
+    "contact_assignment": set(["object_type", "object_id", "contact", "role"]),
     "contact_group": set(["name"]),
     "contact_role": set(["name"]),
     "custom_field": set(["name"]),
@@ -613,6 +616,7 @@
     "console_port_templates": set(["type"]),
     "console_server_ports": set(["type"]),
     "console_server_port_templates": set(["type"]),
+    "contact_assignments": set(["priority"]),
     "devices": set(["status", "face"]),
     "device_types": set(["subdevice_role"]),
     "front_ports": set(["type"]),

plugins/modules/netbox_contact_assignment.py

@@ -0,0 +1,196 @@
+#!/usr/bin/python
+# -*- coding: utf-8 -*-
+# Copyright: (c) 2025, Daniel Chiquito (@dchiquito)
+# GNU General Public License v3.0+ (see COPYING or https://www.gnu.org/licenses/gpl-3.0.txt)
+
+from __future__ import absolute_import, division, print_function
+
+__metaclass__ = type
+
+DOCUMENTATION = r"""
+---
+module: netbox_contact_assignment
+short_description: Creates or removes contact assignments from NetBox
+description:
+  - Creates or removes contact assignments from NetBox
+notes:
+  - Tags should be defined as a YAML list
+  - This should be ran with connection C(local) and hosts C(localhost)
+author:
+  - Daniel Chiquito (@dchiquito)
+requirements:
+  - pynetbox
+version_added: "3.1.0"
+extends_documentation_fragment:
+  - netbox.netbox.common
+options:
+  data:
+    type: dict
+    description:
+      - Defines the contact configuration
+    suboptions:
+      object_type:
+        description:
+          - The type of the object the contact is assigned to
+        required: true
+        type: str
+        choices:
+          - circuit
+          - cluster
+          - cluster_group
+          - device
+          - location
+          - manufacturer
+          - power_panel
+          - provider
+          - rack
+          - region
+          - site
+          - site_group
+          - tenant
+          - virtual_machine
+      object_name:
+        description:
+          - The name of the object the contact is assigned to
+        required: true
+        type: str
+      contact:
+        description:
+          - The name of the contact to assign to the object
+        required: true
+        type: str
+      role:
+        description:
+          - The name of the role the contact has for this object
+        required: true
+        type: str
+      priority:
+        description:
+          - The priority of this contact
+        required: false
+        type: str
+        choices:
+          - primary
+          - secondary
+          - tertiary
+          - inactive
+      tags:
+        description:
+          - Any tags that the contact may need to be associated with
+        required: false
+        type: list
+        elements: raw
+    required: true
+"""
+
+EXAMPLES = r"""
+- name: "Test NetBox module"
+  connection: local
+  hosts: localhost
+  gather_facts: false
+  tasks:
+    - name: Assign a contact to a location with only required information
+      netbox.netbox.netbox_contact_assignment:
+        netbox_url: http://netbox.local
+        netbox_token: thisIsMyToken
+        data:
+          object_type: location
+          object_name: My Location
+          contact: John Doe
+          role: Supervisor Role
+        state: present
+
+    - name: Delete contact assignment within netbox
+      netbox.netbox.netbox_contact_assignment:
+        netbox_url: http://netbox.local
+        netbox_token: thisIsMyToken
+        data:
+          object_type: location
+          object_name: My Location
+          contact: John Doe
+          role: Supervisor Role
+        state: absent
+
+    - name: Create contact with all parameters
+      netbox.netbox.netbox_contact:
+        netbox_url: http://netbox.local
+        netbox_token: thisIsMyToken
+        data:
+          object_type: location
+          object_name: My Location
+          contact: John Doe
+          role: Supervisor Role
+          priority: tertiary
+          tags:
+            - tagA
+            - tagB
+            - tagC
+        state: present
+"""
+
+RETURN = r"""
+contact_assignment:
+  description: Serialized object as created or already existent within NetBox
+  returned: on creation
+  type: dict
+msg:
+  description: Message indicating failure or info about what has been achieved
+  returned: always
+  type: str
+"""
+
+from ansible_collections.netbox.netbox.plugins.module_utils.netbox_utils import (
+    NetboxAnsibleModule,
+    NETBOX_ARG_SPEC,
+)
+from ansible_collections.netbox.netbox.plugins.module_utils.netbox_tenancy import (
+    NetboxTenancyModule,
+    NB_CONTACT_ASSIGNMENTS,
+    OBJECT_TYPES,
+)
+from copy import deepcopy
+
+
+def main():
+    """
+    Main entry point for module execution
+    """
+    argument_spec = deepcopy(NETBOX_ARG_SPEC)
+    argument_spec.update(
+        dict(
+            data=dict(
+                type="dict",
+                required=True,
+                options=dict(
+                    object_type=dict(
+                        required=True, type="str", choices=list(OBJECT_TYPES.keys())
+                    ),
+                    object_name=dict(required=True, type="str"),
+                    contact=dict(required=True, type="str"),
+                    role=dict(required=True, type="str"),
+                    priority=dict(
+                        required=False,
+                        type="str",
+                        choices=["primary", "secondary", "tertiary", "inactive"],
+                    ),
+                    tags=dict(required=False, type="list", elements="raw"),
+                ),
+            ),
+        )
+    )
+
+    required_if = [
+        ("state", "present", ["object_type", "object_name", "contact", "role"]),
+        ("state", "absent", ["object_type", "object_name", "contact", "role"]),
+    ]
+
+    module = NetboxAnsibleModule(
+        argument_spec=argument_spec, supports_check_mode=True, required_if=required_if
+    )
+
+    netbox_contact = NetboxTenancyModule(module, NB_CONTACT_ASSIGNMENTS)
+    netbox_contact.run()
+
+
+if __name__ == "__main__":  # pragma: no cover
+    main()

tests/integration/targets/v4.3/tasks/main.yml

@@ -71,6 +71,15 @@
   tags:
     - netbox_contact
 
+- name: NETBOX_CONTACT_ASSIGNMENT TESTS
+  ansible.builtin.include_tasks:
+    file: netbox_contact_assignment.yml
+    apply:
+      tags:
+        - netbox_contact_assignment
+  tags:
+    - netbox_contact_assignment
+
 - name: NETBOX_CONTACT_ROLE TESTS
   ansible.builtin.include_tasks:
     file: netbox_contact_role.yml