Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
14 changes: 10 additions & 4 deletions log_handler.py
Original file line number Diff line number Diff line change
Expand Up @@ -4,14 +4,20 @@

class LogHandler:
def __new__(cls, *args, **kwargs):
"""
Create and return a new instance of the class.

Returns:
LogHandler: A fresh, uninitialized instance of LogHandler.
"""
return super().__new__(cls)

def __init__(self, args):
"""
Initialize the LogHandler and store the provided arguments.

Initialize the LogHandler with parsed arguments or a configuration object.
Parameters:
args: Parsed arguments or configuration object to be kept on the instance as `self.args`.
args: Parsed command-line arguments or a configuration object; stored on the instance as `self.args`.
"""
self.args = args

Expand Down Expand Up @@ -68,4 +74,4 @@ def log_module_ports_created(self, created_ports: list = [], port_type: str = "p
+ f'{port.type if hasattr(port, "type") else ""} - {port.module_type.id} - '
+ f"{port.id}"
)
return len(created_ports)
return len(created_ports)
38 changes: 33 additions & 5 deletions netbox_api.py
Original file line number Diff line number Diff line change
Expand Up @@ -541,6 +541,16 @@ def create_rear_ports(self, rear_ports, device_type, context=None):
)

def create_front_ports(self, front_ports, device_type, context=None):
"""
Create front port templates for a device type, resolving rear-port references before creation.

For each front-port entry, attempts to resolve its `rear_port` name to the corresponding rear-port ID; front ports whose `rear_port` cannot be resolved are removed and a log entry is emitted (including the optional context). After resolving and pruning entries, the function creates the remaining front-port templates in NetBox and records created items via the shared counters/handlers.

Parameters:
front_ports (list[dict]): List of front-port template definitions. Each item is expected to include a "name" and a "rear_port" (the rear-port name to resolve).
device_type (int): ID of the device type (device_type) to which the front ports belong.
context (str | None): Optional context string appended to log messages for disambiguation.
"""
def link_rear_ports(items, pid):
# Use cached rear ports if available, otherwise fetch from API
cache_key = ("device", pid)
Expand Down Expand Up @@ -681,6 +691,15 @@ def create_module_console_server_ports(self, console_server_ports, module_type,
)

def create_module_rear_ports(self, rear_ports, module_type, context=None):
"""
Create rear-port templates for a module type in NetBox.

Adds any rear port templates from `rear_ports` that do not already exist for the specified `module_type`.
Parameters:
rear_ports (list[dict]): List of rear-port template definitions to create; each item must include a `name` and any other template fields required by NetBox.
module_type (int|object): The module type identifier or object used to associate created templates with the parent module type.
context (str, optional): Optional context string used for logging to identify the source of these templates.
"""
self._create_generic(
rear_ports,
module_type,
Expand All @@ -705,6 +724,15 @@ def create_module_front_ports(self, front_ports, module_type, context=None):

def link_rear_ports(items, pid):
# Use cached rear ports if available, otherwise fetch from API
"""
Resolve each front-port's `rear_port` name to the corresponding rear-port ID for a module and remove any front-ports whose `rear_port` cannot be resolved.

This function mutates `items` in place: for each port dict it replaces the string `rear_port` value with the matching rear-port `.id` when found, logs a message for each missing rear-port (including the available rear-port names), and removes ports with unresolved `rear_port` references. After processing it logs a summary of how many module front ports were skipped. The logs may include the outer-scope `context` value if present.

Parameters:
items (list[dict]): List of front-port dictionaries; each must contain at least `"name"` and `"rear_port"` (the latter initially a name string).
pid (int): The module type ID used to scope the rear-port lookup.
"""
cache_key = ("module", pid)
if "rear_port_templates" in self.cached_components:
existing_rp = self.cached_components["rear_port_templates"].get(cache_key, {})
Expand Down Expand Up @@ -751,10 +779,10 @@ def link_rear_ports(items, pid):

def upload_images(self, baseurl, token, images, device_type):
"""
Upload front and/or rear images to a NetBox device type.

Sends a PATCH request to the device-type endpoint attaching the provided image files, increments self.counter["images"] by the number of files sent, and ensures all opened file handles are closed. The request's SSL verification is controlled by self.ignore_ssl.

Upload front and/or rear image files to the specified NetBox device type.
Sends a PATCH request to the device-type endpoint attaching the provided image files, increments self.counter["images"] by the number of files sent, and ensures all opened file handles are closed. Respects self.ignore_ssl to determine SSL verification behavior.
Parameters:
baseurl (str): Base URL of the NetBox instance (e.g. "https://netbox.example.com").
token (str): API token used for the Authorization header.
Expand Down Expand Up @@ -784,4 +812,4 @@ def upload_images(self, baseurl, token, images, device_type):
finally:
# Ensure all file handles are closed
for _, (_, fh) in file_handles.items():
fh.close()
fh.close()
60 changes: 33 additions & 27 deletions repo.py
Original file line number Diff line number Diff line change
Expand Up @@ -51,13 +51,13 @@ def validate_git_url(url):

def parse_single_file(file):
"""
Load a YAML device file, normalize its manufacturer into a slug dictionary, and add the source path.

Load a YAML device mapping, convert its `manufacturer` to a slug dictionary, and record the source path.
Parameters:
file (str): Path to a YAML file containing a device mapping. The mapping must include a "manufacturer" field.

Returns:
dict: Parsed YAML mapping with `manufacturer` replaced by `{"slug": "<slugified-name>"}` and `src` set to the file path.
dict: Parsed mapping with `manufacturer` replaced by `{"slug": "<slugified-name>"}` and `src` set to the file path.
str: Error string beginning with "Error:" describing YAML parsing or other failure.
"""
with open(file, "r") as stream:
Expand Down Expand Up @@ -87,19 +87,14 @@ def __new__(cls, *args, **kwargs):

def __init__(self, args, repo_path, exception_handler):
"""
Initialize the repository manager and ensure a local clone is present by cloning or pulling.

Sets instance attributes (handler, yaml extensions, URL, repo path, branch, repo reference,
and current working directory). If `repo_path` exists as a directory, updates the repository
by calling `pull_repo` (using the existing remote). Otherwise, validates `args.url` using
`validate_git_url` and creates a local clone by calling `clone_repo`.

Initialize repository management and ensure a local clone exists by either updating an existing clone or cloning the remote.

If the target path already exists as a directory, the repository will be updated from its configured remote; otherwise the provided URL is validated and a new clone is created. The initializer sets instance attributes used by other methods (handler, supported YAML extensions, URL, repo path, branch, repo reference, and current working directory).

Parameters:
args: An object with `url` (str) and `branch` (str) attributes providing the remote
repository URL and branch to use.
repo_path (str): Filesystem path where the repository should be cloned or exists.
exception_handler: An object exposing `exception(name, context, message)` used to
report validation and Git errors as side effects.
args: An object with `url` (str) and `branch` (str) attributes specifying the remote repository URL and branch to use.
repo_path (str): Filesystem path where the repository should be cloned or where an existing clone is located.
exception_handler: An object exposing `exception(name, context, message)` used to report validation and Git errors.
"""
self.handle = exception_handler
self.yaml_extensions = ["yaml", "yml"]
Expand All @@ -121,14 +116,20 @@ def __init__(self, args, repo_path, exception_handler):

def get_relative_path(self):
"""
Return the repository's configured relative path.

Get the repository path configured for this instance relative to the current working directory.
Returns:
str: The instance's stored relative repository path (repo_path).
The stored relative repository path (`repo_path`).
"""
return self.repo_path

def get_absolute_path(self):
"""
Return the absolute filesystem path to the repository directory.

Returns:
str: Absolute path combining the repository path with the repository object's current working directory.
"""
return os.path.join(self.cwd, self.repo_path)

def get_devices_path(self):
Expand Down Expand Up @@ -161,6 +162,11 @@ def pull_repo(self):
self.handle.exception("Exception", "Git Repository Error", git_error)

def clone_repo(self):
"""
Clone the configured Git repository into the configured local path and record the cloned Repo instance.

Attempts to clone from the repository URL into the absolute repository path and set self.repo to the resulting Repo; on success logs the origin URL via the configured handler. If cloning or Git operations fail, the exception is reported to the configured exception handler.
"""
try:
self.repo = Repo.clone_from(self.url, self.get_absolute_path(), branch=self.branch)
self.handle.log(f"Package Installed {self.repo.remotes.origin.url}")
Expand Down Expand Up @@ -199,15 +205,15 @@ def get_devices(self, base_path, vendors: list = None):

def parse_files(self, files: list, slugs: list = None, progress=None):
"""
Parse multiple YAML device files into device type dictionaries, optionally filtering by vendor slugs and integrating with a progress iterator.

Parse YAML device files into device type dictionaries, optionally filtering by vendor slugs and advancing a progress iterable.
Parameters:
files (list): Iterable of file paths to parse.
slugs (list, optional): List of vendor slug substrings to filter results. A parsed item's "slug" is included if any slug from this list is a case-insensitive substring of the item's "slug". If omitted, no slug filtering is applied.
progress (iterable, optional): Optional iterable used to drive a progress display (must yield one item per file). The function iterates this in parallel with parsing so the progress display can be advanced; the values from this iterable are ignored.

files (Iterable[str]): Paths of YAML files to parse.
slugs (list[str], optional): Vendor slug substrings used to filter results; an item is included if any provided slug is a case-insensitive substring of the item's `"slug"`. If omitted, no slug filtering is applied.
progress (Iterable, optional): Iterable consumed in parallel with parsing to drive an external progress display; values are ignored but the iterable should yield once per file.
Returns:
list: A list of parsed device type dictionaries. Files that fail to parse (returned as strings beginning with "Error:") are logged via the instance handler and omitted from the returned list. Files whose parsed data do not match the provided slug filters are also omitted.
list: Parsed device type dictionaries. Files that fail parsing (returned as strings beginning with `"Error:"`) are logged via the instance handler and excluded. Parsed items that do not match the provided slug filters are also excluded.
"""
deviceTypes = []

Expand All @@ -232,4 +238,4 @@ def parse_files(self, files: list, slugs: list = None, progress=None):

deviceTypes.append(data)

return deviceTypes
return deviceTypes