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
8 changes: 7 additions & 1 deletion log_handler.py
Original file line number Diff line number Diff line change
Expand Up @@ -7,6 +7,12 @@ def __new__(cls, *args, **kwargs):
return super().__new__(cls)

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

Parameters:
args: Parsed arguments or configuration object to be kept on the instance as `self.args`.
"""
self.args = args

def exception(self, exception_type, exception, stack_trace=None):
Expand Down Expand Up @@ -62,4 +68,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)
28 changes: 18 additions & 10 deletions netbox_api.py
Original file line number Diff line number Diff line change
Expand Up @@ -685,6 +685,16 @@ def create_module_rear_ports(self, rear_ports, module_type, context=None):
)

def create_module_front_ports(self, front_ports, module_type, context=None):
"""
Create front-port templates for a module-type and link them to their rear ports.

Creates any missing module front-port templates under the given module_type. If a front port references a rear port by name, the rear port name is resolved to the rear-port ID; front ports with unresolved rear-port names are removed from creation and a log message is emitted (includes `context` if provided).

Parameters:
front_ports (list[dict]): List of front-port template definitions. Each dict must include at least "name"; items may reference a rear port by the "rear_port" key (name).
module_type (int | object): Module type identifier or object to associate the created front ports with.
context (str | None): Optional context string appended to log messages for easier debugging.
"""
def link_rear_ports(items, pid):
# Use cached rear ports if available, otherwise fetch from API
cache_key = ("module", pid)
Expand Down Expand Up @@ -726,17 +736,15 @@ def link_rear_ports(items, pid):

def upload_images(self, baseurl, token, images, device_type):
"""
Upload front and/or rear images for a NetBox device type and update the internal image counter.

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.

Parameters:
baseurl (str): Base URL of the NetBox instance (e.g. "https://netbox.example.com").
token (str): API token used for Authorization.
images (dict): Mapping of field name to local file path, e.g. {"front_image": "/path/front.jpg", "rear_image": "/path/rear.jpg"}.
device_type (int | str): Identifier of the device type to update in NetBox.

Notes:
- Increments self.counter["images"] by the number of files successfully sent.
- Ensures file descriptors are closed after the request to avoid resource leaks.
token (str): API token used for the Authorization header.
images (dict): Mapping of form field name to local file path (e.g. {"front_image": "/path/front.jpg", "rear_image": "/path/rear.jpg"}).
device_type (int | str): Identifier of the device type to update in NetBox (used in the endpoint URL).
"""
url = f"{baseurl}/api/dcim/device-types/{device_type}/"
headers = {"Authorization": f"Token {token}"}
Expand All @@ -761,4 +769,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()
52 changes: 42 additions & 10 deletions repo.py
Original file line number Diff line number Diff line change
Expand Up @@ -9,10 +9,14 @@

def validate_git_url(url):
"""
Validate that a git remote URL uses HTTPS or SSH and reject unsafe or local schemes.

Determine whether a Git remote URL is allowed (HTTPS or SSH).

Parameters:
url (str): Git remote URL to validate. Accepted formats are HTTPS URLs with a hostname (e.g., https://host/...),
SSH scp-like form beginning with `git@host:` (e.g., git@host:org/repo.git), or ssh URLs starting with `ssh://`.

Returns:
tuple: (bool, str or None) — (True, None) if the URL is allowed; (False, error_message) otherwise.
(bool, str or None): `True, None` if the URL is allowed; otherwise `False` and a short error message explaining why.
"""
if not url or not str(url).strip():
return False, "Empty URL"
Expand Down Expand Up @@ -47,14 +51,14 @@ def validate_git_url(url):

def parse_single_file(file):
"""
Load a YAML device file, replace its "manufacturer" value with a slug dictionary, add the source path, and return the parsed data or an error string.

Load a YAML device file, normalize its manufacturer into a slug dictionary, and add the source path.
Parameters:
file (str): Path to a YAML file containing device data. The file must include a "manufacturer" field.

file (str): Path to a YAML file containing a device mapping. The mapping must include a "manufacturer" field.
Returns:
dict: The parsed YAML mapping with "manufacturer" replaced by {"slug": "<slugified-name>"} and "src" set to the file path.
str: An error string beginning with "Error:" describing YAML parsing or other failures.
dict: Parsed YAML 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:
try:
Expand All @@ -73,6 +77,12 @@ def parse_single_file(file):

class DTLRepo:
def __new__(cls, *args, **kwargs):
"""
Allocate and return a new instance of the class using the default object allocator.

Returns:
instance: A newly created instance of the class `cls`.
"""
return super().__new__(cls)

def __init__(self, args, repo_path, exception_handler):
Expand Down Expand Up @@ -110,6 +120,12 @@ def __init__(self, args, repo_path, exception_handler):
self.clone_repo()

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

Returns:
str: The instance's stored relative repository path (repo_path).
"""
return self.repo_path

def get_absolute_path(self):
Expand Down Expand Up @@ -154,6 +170,22 @@ def clone_repo(self):
self.handle.exception("Exception", "Git Repository Error", git_error)

def get_devices(self, base_path, vendors: list = None):
"""
Discover device YAML files and vendor directories under a base path.

Parameters:
base_path (str): Directory path containing vendor subdirectories (each vendor folder contains device YAML files).
vendors (list, optional): List of vendor names (case-insensitive) to include; if omitted, all vendors are considered.

Returns:
tuple:
files (list): List of file paths to discovered YAML files (extensions from self.yaml_extensions) under matching vendor folders.
discovered_vendors (list): List of dicts for each discovered vendor with keys:
- name (str): Vendor directory name.
- slug (str): Slugified vendor name produced by self.slug_format.
Note:
The folder named "testing" (case-insensitive) is ignored.
"""
files = []
discovered_vendors = []
vendor_dirs = os.listdir(base_path)
Expand Down Expand Up @@ -200,4 +232,4 @@ def parse_files(self, files: list, slugs: list = None, progress=None):

deviceTypes.append(data)

return deviceTypes
return deviceTypes