Skip to main content

configuration

switchmap classes that manage various configurations.

ConfigServer Objects​

class ConfigServer(ConfigAPI)

Class gathers all configuration information.

__init__​

def __init__()

Intialize the class.

Arguments:

None

Returns:

None

api_bind_port​

def api_bind_port()

Get api_bind_port.

Arguments:

None

Returns:

  • result - result

cache_directory​

def cache_directory()

Determine the cache_directory.

Arguments:

None

Returns:

  • result - configured cache_directory

db_host​

def db_host()

Return db_host value.

Arguments:

None

Returns:

  • result - db_host value

db_name​

def db_name()

Return db_name value.

Arguments:

None

Returns:

  • result - db_name value

db_max_overflow​

def db_max_overflow()

Get DB connection pool overflow size.

Arguments:

None

Returns:

  • result - Configured value

db_pass​

def db_pass()

Return db_pass value.

Arguments:

None

Returns:

  • result - db_pass value

db_pool_size​

def db_pool_size()

Get DB connection pool size.

Arguments:

None

Returns:

  • result - Configured value

db_user​

def db_user()

Return db_user value.

Arguments:

None

Returns:

  • result - db_user value

ingest_directory​

def ingest_directory()

Determine the ingest_directory.

Arguments:

None

Returns:

  • result - ingest_directory

ingest_interval​

def ingest_interval()

Get ingest_interval.

Arguments:

None

Returns:

  • result - result

purge_after_ingest​

def purge_after_ingest()

Return purge_after_ingest value.

Arguments:

None

Returns:

  • result - purge_after_ingest value

username​

def username()

Get username.

Arguments:

None

Returns:

  • result - result

__init__

Define the switchmap.server package.

Arguments:

None

Returns:

None

db

Module to manage connection pooling among other things.

main​

def main()

Process agent data.

Arguments:

None

Returns:

None

db.ingest.ingest

switchmap classes that manage the DB uploading of polled data.

Ingest Objects​

class Ingest()

Read cache files in the DB.

__init__​

def __init__(config,
             test=False,
             test_cache_directory=None,
             multiprocessing=False)

Initialize class.

Arguments:

  • config - ConfigServer object
  • test - True if testing
  • test_cache_directory - Ingest directory. Only used when testing.
  • multiprocessing - True if multiprocessing is enabled

Returns:

None

process​

def process()

Process files in the cache.

Arguments:

None

Returns:

None

zone​

def zone(arguments)

Ingest the files' zone data.

Arguments:

  • arguments - List of Argument objects

Returns:

  • success - True if successful

device​

def device(arguments)

Ingest the files' device data.

Arguments:

  • arguments - List of arguments for the processing the zone [[item.idx_zone, item.data, item.filepath, item.config]]

Returns:

  • success - True if successful

cleanup​

def cleanup(event)

Ingest the files' device data.

Arguments:

  • event - Name of event

Returns:

None

process_zone​

def process_zone(argument)

Ingest a single file for device updates.

Arguments:

  • argument - Argument object

Returns:

  • rows - ZoneObjects object

process_device​

def process_device(argument)

Ingest a single file for device updates.

Arguments:

  • argument - Argument object

Returns:

None

setup​

def setup(src, config)

Ingest the files in parallel.

Arguments:

  • src - Directory where device YAML files are located
  • config - Configuration object

Returns:

  • result - EventObjects object

insert_arptable​

def insert_arptable(data, test=False)

Insert values from ARP tables.

Arguments:

  • data - List of lists of ZoneObjects, one per device OR a single ZoneObjects from testing
  • test - Sequentially insert values into the database if True. Bulk inserts don't insert data with predictable primary keys.

Returns:

  • pairmacips - List of PairMacIp objects

insert_macips​

def insert_macips(items, test=False)

Update the mac DB table.

Arguments:

  • items - List of PairMacIp objects
  • test - Sequentially insert values into the database if True. Bulk inserts don't insert data with predictable primary keys.

Returns:

None

db.ingest.query.device

Module to check the Device table when updating the DB with polled data.

Device Objects​

class Device()

Get all Device data.

__init__​

def __init__(idx_zone, hostname)

Initialize class.

Arguments:

  • idx_zone - Zone index to which the data belongs
  • hostname - Hostname to process

Returns:

None

data​

def data()

Return complete device.

Arguments:

None

Returns:

  • Result - DeviceDetail object

device​

def device()

Return system summary data.

Arguments:

None

Returns:

  • Result - RDevice object

interfaces​

def interfaces()

Return L1 data for Ethernet ports only.

Arguments:

None

Returns:

  • self._ports - L1 data for Ethernet ports

vlanports​

def vlanports(idx_device)

Get all the VlanPorts for a device.

Arguments:

  • idx_device - Idx_device of the device being processed

Returns:

  • result - List of RVlanPort tuple

db.ingest.query.mac

Module for getting interface specific mac data.

by_idx_mac​

def by_idx_mac(idx_mac)

Search for MAC addresses.

Arguments:

  • idx_mac - idx_mac

Returns:

  • result - List of MacDetail objects

db.ingest.query.vlan

Module for getting interface specific VLAN data.

by_idx_l1interface​

def by_idx_l1interface(idx_l1interface)

Search for VLANs tied to interface.

Arguments:

  • idx_l1interface - idx_l1interface

Returns:

  • result - List of RVlan objects

db.ingest.query

Define the switchmap.server.db.ingest.query package.

Arguments:

None

Returns:

None

db.ingest

Define the switchmap.server.db.ingest package.

Arguments:

None

Returns:

None

db.ingest.update.device

Module for updating the database with topology data.

process​

def process(data, idx_zone, dns=True)

Process data received from a device.

Arguments:

  • data - Device data (dict)
  • idx_zone - Zone index to which the data belongs
  • dns - Do DNS lookups if True

Returns:

None

device​

def device(idx_zone, data)

Update the Device DB table.

Arguments:

  • idx_zone - Zone index to which the data belongs
  • data - Device data (dict)

Returns:

None

Status Objects​

class Status()

Tracks the status of various Topology methods.

__init__​

def __init__()

Instantiate the class.

Arguments:

None

Returns:

None

l1interface​

@property
def l1interface()

Provide the value of the 'l1interface' property.

l1interface​

@l1interface.setter
def l1interface(value)

Set the 'l1interface' property.

systemstat​

@property
def systemstat()

Provide the value of the 'systemstat' property.

systemstat​

@systemstat.setter
def systemstat(value)

Set the 'systemstat' property.

ipport​

@property
def ipport()

Provide the value of the 'ipport' property.

ipport​

@ipport.setter
def ipport(value)

Set the 'ipport' property.

macport​

@property
def macport()

Provide the value of the 'macport' property.

macport​

@macport.setter
def macport(value)

Set the 'macport' property.

vlanport​

@property
def vlanport()

Provide the value of the 'vlanport' property.

vlanport​

@vlanport.setter
def vlanport(value)

Set the 'vlanport' property.

vlan​

@property
def vlan()

Provide the value of the 'vlan' property.

vlan​

@vlan.setter
def vlan(value)

Set the 'vlan' property.

Topology Objects​

class Topology()

Update Device data in the database.

__init__​

def __init__(exists, data, dns=True)

Initialize class.

Arguments:

  • exists - RDevice object
  • data - Dict of device data
  • dns - Do DNS lookups if True

Returns:

None

process​

def process()

Process data received from a device.

Arguments:

None

Returns:

None

l1interface​

def l1interface(test=False)

Update the L1interface DB table.

Arguments:

  • test - Sequentially insert values into the database if True. Bulk inserts don't insert data with predictable primary keys.

Returns:

None

vlan​

def vlan(test=False)

Update the Vlan DB table.

Arguments:

  • test - Sequentially insert values into the database if True. Bulk inserts don't insert data with predictable primary keys.

Returns:

None

vlanport​

def vlanport(test=False)

Update the VlanPort DB table.

Arguments:

  • test - Sequentially insert values into the database if True. Bulk inserts don't insert data with predictable primary keys.

Returns:

None

macport​

def macport(test=False)

Update the MacPort DB table.

Arguments:

  • test - Sequentially insert values into the database if True. Bulk inserts don't insert data with predictable primary keys.

Returns:

None

ipport​

def ipport(test=False)

Update the IpPort DB table.

Arguments:

  • test - Sequentially insert values into the database if True. Bulk inserts don't insert data with predictable primary keys.

Returns:

None

systemstat​

def systemstat(test=False)

Update the SystemStat DB table.

Arguments:

  • test - Test mode if True

Returns:

None

log​

def log(table, updated=False)

Create standardized log messaging.

Arguments:

  • table - Name of table being updated
  • updated - True if the table has been updated

Returns:

None

log_invalid​

def log_invalid(table)

Create standardized log messaging for invalid states.

Arguments:

  • table - Name of table being updated

Returns:

None

db.ingest.update

Define the switchmap.server.db.ingest.update package.

Arguments:

None

Returns:

None

db.ingest.update.zone

Module for updating the database with topology data.

process​

def process(data, idx_zone, dns=True)

Process data received from a device.

Arguments:

  • data - Device data (dict)
  • idx_zone - Zone index to which the data belongs
  • dns - Do DNS lookups if True

Returns:

  • results - ZoneObjects object

Status Objects​

class Status()

Tracks the status of various Topology methods.

__init__​

def __init__()

Instantiate the class.

Arguments:

None

Returns:

None

ip​

@property
def ip()

Provide the value of the 'ip' property.

Arguments:

None

Returns:

None

ip​

@ip.setter
def ip(value)

Set the 'ip' property.

Arguments:

  • value - Value to set

Returns:

None

mac​

@property
def mac()

Provide the value of the 'mac' property.

Arguments:

None

Returns:

None

mac​

@mac.setter
def mac(value)

Set the 'mac' property.

Arguments:

  • value - Value to set

Returns:

None

macip​

@property
def macip()

Provide the value of the 'macip' property.

Arguments:

None

Returns:

None

macip​

@macip.setter
def macip(value)

Set the 'macip' property.

Arguments:

  • value - Value to set

Returns:

None

Topology Objects​

class Topology()

Update Device data in the database.

__init__​

def __init__(data, idx_zone, dns=True)

Initialize class.

Arguments:

  • data - Dict of device data
  • idx_zone - idx_zone of the Zone being processed
  • dns - Do DNS lookups if True

Returns:

None

process​

def process()

Process data received from a device.

Arguments:

None

Returns:

None

mac​

def mac()

Update the Mac DB table.

Arguments:

None

Returns:

  • rows - List of unique IMac objects

ip​

def ip()

Update the Ip DB table.

Arguments:

None

Returns:

None

macip​

def macip()

Update the MacIp DB table.

Arguments:

None

Returns:

  • rows - List of PairMacIp objects

log​

def log(table, updated=False)

Create standardized log messaging.

Arguments:

  • table - Name of table being updated
  • updated - True if the table has been updated

Returns:

None

log_invalid​

def log_invalid(table)

Create standardized log messaging for invalid states.

Arguments:

  • table - Name of table being updated

Returns:

None

db.models

Define SQLalchemy database table models.

Oui Objects​

class Oui(BASE)

Database table definition.

Event Objects​

class Event(BASE)

Database table definition.

Root Objects​

class Root(BASE)

Database table definition.

Zone Objects​

class Zone(BASE)

Database table definition.

Device Objects​

class Device(BASE)

Database table definition.

SystemStat Objects​

class SystemStat(BASE)

Database table definition.

L1Interface Objects​

class L1Interface(BASE)

Database table definition.

Vlan Objects​

class Vlan(BASE)

Database table definition.

VlanPort Objects​

class VlanPort(BASE)

Database table definition.

Mac Objects​

class Mac(BASE)

Database table definition.

MacPort Objects​

class MacPort(BASE)

Database table definition.

Ip Objects​

class Ip(BASE)

Database table definition.

IpPort Objects​

class IpPort(BASE)

Database table definition.

MacIp Objects​

class MacIp(BASE)

Database table definition.

create_all_tables​

def create_all_tables()

Ensure all tables are created.

Arguments:

None

Returns:

None

db.db

Class to process connection.

db_select_row​

def db_select_row(error_code, statement)

Support 'Select' actions for ENTIRE row.

Arguments:

  • error_code - Error code to use in messages
  • statement - SqlALchemy statement to execute. This must only reference an ORM Row object.

Returns:

db_select​

def db_select(error_code, statement)

Provide a transactional support for Select actions.

Arguments:

  • error_code - Error code to use in messages
  • statement - SqlALchemy statement to execute

Returns:

db_update​

def db_update(error_code, statement, values=None)

Provide a transactional support for Update actions.

Arguments:

  • error_code - Error code to use in messages
  • statement - SqlALchemy statement to execute
  • values - List of values to insert if required

Returns:

db_delete_row​

def db_delete_row(error_code, statement)

Support 'Delete' actions for ENTIRE row.

Arguments:

  • error_code - Error code to use in messages
  • statement - SqlALchemy statement to execute. This must only reference an ORM Row object.

Returns:

db_delete​

def db_delete(error_code, statement)

Provide a transactional support for Delete actions.

Arguments:

  • error_code - Error code to use in messages
  • statement - SqlALchemy statement to execute

Returns:

  • result - Number of affected rows

db_insert_row​

def db_insert_row(error_code, model, mappings, die=True)

Perform bulk insert for ORM objects with enhanced logging.

Arguments:

  • error_code - Error code to use in messages
  • model - SQLAlchemy model to insert into
  • mappings - List of dictionaries with data to insert
  • die - Die if True

Returns:

  • result - True if successful

db.schemas

ORM Schema classes.

Used for defining GraphQL interaction

Based on the pages at:

https://github.com/alexisrolland/flask-graphene-sqlalchemy/wiki/Flask-Graphene-SQLAlchemy-Tutorial https://docs.graphene-python.org/projects/sqlalchemy/en/latest/tutorial/

Event Objects​

class Event(SQLAlchemyObjectType, EventAttribute)

Event node.

Meta Objects​

class Meta()

Define the metadata.

Root Objects​

class Root(SQLAlchemyObjectType, RootAttribute)

Root node.

Meta Objects​

class Meta()

Define the metadata.

Device Objects​

class Device(SQLAlchemyObjectType, DeviceAttribute)

Device node.

Meta Objects​

class Meta()

Define the metadata.

SystemStat Objects​

class SystemStat(SQLAlchemyObjectType, SystemStatAttribute)

SystemStat node.

Meta Objects​

class Meta()

Define the metadata.

Ip Objects​

class Ip(SQLAlchemyObjectType, IpAttribute)

Ip node.

Meta Objects​

class Meta()

Define the metadata.

IpPort Objects​

class IpPort(SQLAlchemyObjectType, IpPortAttribute)

IpPort node.

Meta Objects​

class Meta()

Define the metadata.

L1Interface Objects​

class L1Interface(SQLAlchemyObjectType, L1InterfaceAttribute)

L1Interface node.

Meta Objects​

class Meta()

Define the metadata.

Mac Objects​

class Mac(SQLAlchemyObjectType, MacAttribute)

Mac node.

Meta Objects​

class Meta()

Define the metadata.

Zone Objects​

class Zone(SQLAlchemyObjectType, ZoneAttribute)

Zone node.

Meta Objects​

class Meta()

Define the metadata.

MacIp Objects​

class MacIp(SQLAlchemyObjectType, MacIpAttribute)

MacIp node.

Meta Objects​

class Meta()

Define the metadata.

MacPort Objects​

class MacPort(SQLAlchemyObjectType, MacPortAttribute)

MacPort node.

Meta Objects​

class Meta()

Define the metadata.

Oui Objects​

class Oui(SQLAlchemyObjectType, OuiAttribute)

Oui node.

Meta Objects​

class Meta()

Define the metadata.

Vlan Objects​

class Vlan(SQLAlchemyObjectType, VlanAttribute)

Vlan node.

Meta Objects​

class Meta()

Define the metadata.

VlanPort Objects​

class VlanPort(SQLAlchemyObjectType, VlanPortAttribute)

VlanPort node.

Meta Objects​

class Meta()

Define the metadata.

Query Objects​

class Query(graphene.ObjectType)

Define GraphQL queries.

resolve_devices​

def resolve_devices(root, info, hostname=None, **kwargs)

Resolve and return devices from the database.

Arguments:

  • root - The root object (not used here).
  • info - GraphQL resolver info, used to get the query context.
  • hostname str, optional - If provided, filters by this hostname.
  • **kwargs - Additional arguments (ignored).

Returns:

  • sqlalchemy.orm.Query - A query object for the matching Device.

db.table.ipport

Module for querying the IpPort table.

idx_exists​

def idx_exists(idx)

Determine whether primary key exists.

Arguments:

  • idx - idx_ipport

Returns:

  • result - True if exists

exists​

def exists(idx_l1interface, idx_ip)

Determine whether entry exists in the IpPort table.

Arguments:

  • idx_l1interface - Device.idx_l1interface
  • idx_ip - Ip.idx_ip

Returns:

  • result - RIpPort tuple

find_idx_ip​

def find_idx_ip(idx_ip)

Find all ports on which MAC address has been found.

Arguments:

  • idx_ip - Ip.idx_ip

Returns:

  • result - RIpPort tuple

insert_row​

def insert_row(rows)

Create a IpPort table entry.

Arguments:

  • rows - IIpPort objects

Returns:

None

update_row​

def update_row(idx, row)

Upadate a IpPort table entry.

Arguments:

  • idx - idx_ipport value
  • row - IIpPort object

Returns:

None

db.table.device

Module for querying the Device table.

idx_exists​

def idx_exists(idx)

Determine whether primary key exists.

Arguments:

  • idx - idx_device

Returns:

  • result - RDevice object

exists​

def exists(idx_zone, hostname)

Determine whether hostname exists in the Device table.

Arguments:

  • idx_zone - Zone index
  • hostname - Device

Returns:

  • result - RDevice tuple

devices​

def devices(idx_zone)

Get all Devices for a zone.

Arguments:

  • idx_zone - Zone index

Returns:

  • result - list of RDevice tuples

insert_row​

def insert_row(rows)

Create a Device table entry.

Arguments:

  • rows - IDevice objects

Returns:

None

update_row​

def update_row(idx, row)

Upadate a Device table entry.

Arguments:

  • idx - idx_device value
  • row - IDevice object

Returns:

None

db.table.systemstat

Module for querying the SystemStat table.

idx_exists​

def idx_exists(idx)

Determine whether primary key exists.

Arguments:

  • idx - idx_systemstat

Returns:

  • results - SystemStat object

device_exists​

def device_exists(idx_device)

Determine whether SystemStat record exists for a device.

Arguments:

  • idx_device - Device index

Returns:

  • result - SystemStat object or False if not found

insert_row​

def insert_row(rows)

Create a SystemStat table entry.

Arguments:

  • rows - SystemStat objects

Returns:

None

update_row​

def update_row(idx, row)

Update a systemstat table entry.

Arguments:

  • idx - idx_systemstat value
  • row - SystemStat object

Returns:

None

db.table.l1interface

Module for querying the L1Interface table.

idx_exists​

def idx_exists(idx)

Determine whether primary key exists.

Arguments:

  • idx - idx_l1interface

Returns:

  • result - RL1Interface object

exists​

def exists(idx_device, ifindex)

Determine whether hostname exists in the L1Interface table.

Arguments:

  • idx_device - Device.idx_device
  • ifindex - SNMP IfIndex number

Returns:

  • result - RL1Interface tuple

ifindexes​

def ifindexes(idx_device)

Get all the L1Interface table records for a device.

Arguments:

  • idx_device - Device.idx_device

Returns:

  • result - RL1Interface tuple

findifalias​

def findifalias(idx_device, ifalias)

Find ifalias.

Arguments:

  • idx_device - Device.idx_device
  • ifalias - Hostname

Returns:

  • idx_device - Device.idx_device
  • result - list of L1Interface tuples

insert_row​

def insert_row(rows)

Create a L1Interface table entry.

Arguments:

  • rows - IL1Interface objects

Returns:

None

update_row​

def update_row(idx, row)

Update a L1Interface table entry.

Arguments:

  • idx - idx_l1interface value
  • row - IL1Interface object

Returns:

None

db.table.vlanport

Module for querying the VlanPort table.

idx_exists​

def idx_exists(idx)

Determine whether primary key exists.

Arguments:

  • idx - idx_vlanport

Returns:

  • result - True if exists

exists​

def exists(idx_l1interface, idx_vlan)

Determine whether entry exists in the VlanPort table.

Arguments:

  • idx_l1interface - Device.idx_l1interface
  • idx_vlan - Vlan.idx_vlan

Returns:

  • result - RVlanPort tuple

find_idx_vlan​

def find_idx_vlan(idx_vlan)

Find all ports on which MAC address has been found.

Arguments:

  • idx_vlan - Vlan.idx_vlan

Returns:

  • result - RVlanPort tuple

insert_row​

def insert_row(rows)

Create a VlanPort table entry.

Arguments:

  • rows - IVlanPort objects

Returns:

None

update_row​

def update_row(idx, row)

Upadate a VlanPort table entry.

Arguments:

  • idx - idx_vlanport value
  • row - IVlanPort object

Returns:

None

db.table.macport

Module for querying the MacPort table.

idx_exists​

def idx_exists(idx)

Determine whether primary key exists.

Arguments:

  • idx - idx_macport

Returns:

  • result - True if exists

exists​

def exists(idx_l1interface, idx_mac)

Determine whether entry exists in the MacPort table.

Arguments:

  • idx_l1interface - Device.idx_l1interface
  • idx_mac - Mac.idx_mac

Returns:

  • result - RMacPort tuple

find_idx_mac​

def find_idx_mac(idx_mac)

Find all ports on which MAC address has been found.

Arguments:

  • idx_mac - Mac.idx_mac

Returns:

  • result - RMacPort tuple

insert_row​

def insert_row(rows)

Create a MacPort table entry.

Arguments:

  • rows - IMacPort objects

Returns:

None

update_row​

def update_row(idx, row)

Upadate a MacPort table entry.

Arguments:

  • idx - idx_macport value
  • row - IMacPort object

Returns:

None

db.table.mac

Module for querying the Mac table.

idx_exists​

def idx_exists(idx)

Determine whether primary key exists.

Arguments:

  • idx - idx_mac

Returns:

  • result - RMac object

exists​

def exists(idx_zone, _mac)

Determine whether MAC exists in the Mac table.

Arguments:

  • idx_zone - Zone index
  • _mac - Mac address

Returns:

  • result - RMac tuple

findmac​

def findmac(idx_zone, macs)

Determine whether MAC exists in the Mac table.

Arguments:

  • idx_zone - Zone index
  • macs - List or single MAC address

Returns:

  • result - list of RMac tuples

insert_row​

def insert_row(rows)

Create a Mac table entry.

Arguments:

  • rows - TopologyMac objects

Returns:

None

update_row​

def update_row(idx, row)

Upadate a Mac table entry.

Arguments:

  • idx - idx_mac value
  • row - IMac object

Returns:

None

db.table.ip

Module for querying the Ip table.

idx_exists​

def idx_exists(idx)

Determine whether primary key exists.

Arguments:

  • idx - idx_ip

Returns:

  • result - RIp object

exists​

def exists(idx_zone, _ip)

Determine whether MAC exists in the Ip table.

Arguments:

  • idx_zone - Zone index
  • _ip - Ip address

Returns:

  • result - RIp tuple

findhostname​

def findhostname(idx_zone, hostname)

Determine whether hostname exists in the Ip table.

Arguments:

  • idx_zone - Zone index
  • hostname - hostname

Returns:

  • results - RIp list

findip​

def findip(idx_zone, ips)

Determine whether MAC exists in the Ip table.

Arguments:

  • idx_zone - Zone index
  • ips - one or more IP addresses

Returns:

  • result - list of RIp tuples

insert_row​

def insert_row(rows)

Perform bulk insert for the Mac table.

Arguments:

  • rows - List of IMac objects to be inserted into the database.

Returns:

  • None - This function does not return any value.

update_row​

def update_row(idx, row)

Upadate a Ip table entry.

Arguments:

  • idx - idx_ip value
  • row - IIp object

Returns:

None

db.table.vlan

Module for querying the Vlan table.

idx_exists​

def idx_exists(idx)

Determine whether primary key exists.

Arguments:

  • idx - idx_vlan

Returns:

  • result - RVlan object

exists​

def exists(idx_device, vlan)

Determine whether vlan exists in the Vlan table.

Arguments:

  • idx_device - DB idx for the device
  • vlan - Vlan

Returns:

  • result - RVlan tuple

vlans​

def vlans(idx_device)

Get all VLANs for a device.

Arguments:

  • idx_device - Device index

Returns:

  • result - list of RVlan tuples

insert_row​

def insert_row(rows)

Create a Vlan table entry.

Arguments:

  • rows - IVlan objects

Returns:

None

update_row​

def update_row(idx, row)

Upadate a Vlan table entry.

Arguments:

  • idx - idx_vlan value
  • row - IVlan object

Returns:

None

db.table.root

Module for querying the Root table.

idx_exists​

def idx_exists(idx)

Determine whether primary key exists.

Arguments:

  • idx - idx_root

Returns:

  • result - RRoot record

exists​

def exists(root)

Determine whether root exists in the Root table.

Arguments:

  • root - Root

Returns:

  • result - RRoot tuple

insert_row​

def insert_row(rows)

Create a Root table entry.

Arguments:

  • rows - IRoot objects

Returns:

None

update_row​

def update_row(idx, row)

Upadate a Root table entry.

Arguments:

  • idx - idx_root value
  • row - IRoot object

Returns:

None

roots​

def roots()

Get list of Roots.

Arguments:

None

Returns:

  • result - RRoot object

db.table

Constants required for DB queries and updates.

db.table.event

Module for querying the Event table.

idx_exists​

def idx_exists(idx)

Determine whether primary key exists.

Arguments:

  • idx - idx_event

Returns:

  • result - REvent object

exists​

def exists(event)

Determine whether event exists in the Event table.

Arguments:

  • event - Event

Returns:

  • result - REvent tuple

insert_row​

def insert_row(rows)

Create a Event table entry.

Arguments:

  • rows - IEvent objects

Returns:

None

update_row​

def update_row(idx, row)

Upadate a Event table entry.

Arguments:

  • idx - idx_event value
  • row - IEvent object

Returns:

None

events​

def events()

Get list of Events.

Arguments:

None

Returns:

  • result - REvent object

delete​

def delete(idx)

Delete event.

Arguments:

  • idx - idx_event

Returns:

None

create​

def create(name=None)

Create an event.

Arguments:

  • name - Alternative name

Returns:

  • result - Event object for row that doesn't already exist

purge​

def purge()

Purge all events except the most recent two.

Arguments:

None

Returns:

  • result - None

db.table.zone

Module for querying the Zone table.

idx_exists​

def idx_exists(idx)

Determine whether primary key exists.

Arguments:

  • idx - idx_zone

Returns:

  • result - RZone record

exists​

def exists(idx_event, name)

Determine whether name exists in the Zone table.

Arguments:

  • idx_event - Event index
  • name - Zone

Returns:

  • result - RZone tuple

zones​

def zones(idx_event)

Get all Zones for a event.

Arguments:

  • idx_event - Event index

Returns:

  • result - list of RZone tuples

insert_row​

def insert_row(rows)

Create a Zone table entry.

Arguments:

  • rows - IZone objects

Returns:

None

update_row​

def update_row(idx, row)

Upadate a Zone table entry.

Arguments:

  • idx - idx_zone value
  • row - IZone object

Returns:

None

db.table.macip

Module for querying the MacIp table.

idx_exists​

def idx_exists(idx)

Determine whether primary key exists.

Arguments:

  • idx - idx_macip

Returns:

  • result - RMacIp object

exists​

def exists(idx_mac, idx_ip)

Determine whether mac exists in the MacIp table.

Arguments:

  • idx_mac - Mac.idx_mac
  • idx_ip - Ip.idx_ip

Returns:

  • result - RMacIp tuple

insert_row​

def insert_row(rows)

Create a MacIp table entry.

Arguments:

  • rows - IMacIp objects

Returns:

None

update_row​

def update_row(idx, row)

Upadate a MacIp table entry.

Arguments:

  • idx - idx_macip value
  • row - IMacIp object

Returns:

None

idx_ips_exist​

def idx_ips_exist(idx_mac)

Get a list of idx_ip values matching idx_mac from the MacIp table.

Arguments:

  • idx_mac - Mac.idx_mac

Returns:

  • result - List of RMacIp tuples

db.table.oui

Module for querying the Oui table.

idx_oui​

def idx_oui(mac)

Get the idx_oui value.

Arguments:

  • mac - MAC address

Returns:

  • result - idx_oui value

idx_exists​

def idx_exists(idx)

Determine whether primary key exists.

Arguments:

  • idx - idx_oui

Returns:

  • result - ROui record

exists​

def exists(oui)

Determine whether oui exists in the Oui table.

Arguments:

  • oui - Oui

Returns:

  • result - ROui tuple

insert_row​

def insert_row(rows)

Create a Oui table entry.

Arguments:

  • rows - IOui objects

Returns:

None

update_row​

def update_row(idx, row)

Upadate a Oui table entry.

Arguments:

  • idx - idx_oui value
  • row - IOui object

Returns:

None

ouis​

def ouis()

Get all the OUIs.

Arguments:

None

Returns:

  • result - ROui record

db.misc.rows

Module to handle database table rows.

device​

def device(row)

Convert table row to tuple.

Arguments:

  • row - Device row

Returns:

  • result - RDevice tuple

root​

def root(row)

Convert table row to tuple.

Arguments:

  • row - Root row

Returns:

  • result - RRoot tuple

event​

def event(row)

Convert table row to tuple.

Arguments:

  • row - Event row

Returns:

  • result - REvent tuple

systemstat​

def systemstat(row)

Convert table row to tuple.

Arguments:

  • row - SystemStat row

Returns:

  • results - SystemStat tuple

l1interface​

def l1interface(row)

Convert table row to tuple.

Arguments:

  • row - L1Interface row

Returns:

  • result - RL1Interface tuple

mac​

def mac(row)

Convert table row to tuple.

Arguments:

  • row - Mac row

Returns:

  • result - RMac tuple

macip​

def macip(row)

Convert table row to tuple.

Arguments:

  • row - MacIp row

Returns:

  • result - RMacIp tuple

macport​

def macport(row)

Convert table row to tuple.

Arguments:

  • row - MacPort row

Returns:

  • result - RMacPort tuple

oui​

def oui(row)

Convert table row to tuple.

Arguments:

  • row - Oui row

Returns:

  • result - ROui tuple

vlan​

def vlan(row)

Convert table row to tuple.

Arguments:

  • row - Vlan row

Returns:

  • result - RVlan tuple

vlanport​

def vlanport(row)

Convert table row to tuple.

Arguments:

  • row - VlanPort row

Returns:

  • result - RVlanPort tuple

zone​

def zone(row)

Convert table row to tuple.

Arguments:

  • row - Zone row

Returns:

  • result - RZone tuple

ip​

def ip(row)

Convert table row to tuple.

Arguments:

  • row - Ip row

Returns:

  • result - RIp tuple

ipport​

def ipport(row)

Convert table row to tuple.

Arguments:

  • row - IpPort row

Returns:

  • result - RIpPort tuple

db.misc.interface

Switchmap Interface library.

interfaces​

def interfaces(rdevice)

Get an Rl1interface list for the device during the previous event.

Arguments:

  • rdevice - RDevice object

Returns:

  • result - List of matching Rl1interface objects

db.misc.search

switchmap.Search class.

Description:

This files has classes that process searches for:

  1. IP and MAC address
  2. Port names
  3. Hostnames

Search Objects​

class Search()

Class that manages searches.

Methods return lists of Found objects for the idx_l1interface table entries where there are matches.

__init__​

def __init__(idx_event, searchstring)

Initialize the class.

Arguments:

  • idx_event - Event index
  • searchstring - search string to look for

Returns:

None

find​

def find()

Find search string.

Arguments:

None

Returns:

  • result - List of Found objects of interfaces that have data matching the search string

macaddress​

def macaddress()

Search for macaddress.

Arguments:

None

Returns:

  • result - List of Found objects of interfaces that have data matching the search string

ifalias​

def ifalias()

Search for string in ifalias names.

Arguments:

None

Returns:

  • result - List of Found objects of interfaces that have data matching the search string

ipaddress​

def ipaddress()

Search for ipaddress.

Arguments:

None

Returns:

  • result - List of Found objects of interfaces that have data matching the search string

hostname​

def hostname()

Search for string hostnames.

Arguments:

None

Returns:

  • result - List of Found objects of interfaces that have data matching the search string

find_ip_interface​

def find_ip_interface(idx_ip)

Find all ports on which an specfic IP address has been found.

Arguments:

  • idx_ip - Ip.idx_ip

Returns:

  • result - RIpPort tuple
def search(idx_root, searchstring)

Search based on idx_root values.

Arguments:

  • idx_root - Root index
  • searchstring - search string to look for

Returns:

  • result - List of idx_l1interface values

trunk​

def trunk(idx_l1interface)

Determine whether the interface is a trunk.

Arguments:

  • idx_l1interface - L1interface table primary key

Returns:

  • result - True / False status

db.misc

Define the switchmap.server.db.misc package.

Arguments:

None

Returns:

None

db.misc.oui

Switchmap OUI library.

update_db_oui​

def update_db_oui(filepath, new=False)

Update the database with Oui data.

Arguments:

  • filepath - File to process
  • new - If True, skip existing entry checks for new installations

Returns:

None

Raises:

  • FileNotFoundError - If the file cannot be found
  • ValueError - If the CSV is improperly formatted

db.attributes

ORM attribute classes for SQLAlchemy graphene / GraphQL schemas..

Used for defining GraphQL interaction

Based on the pages at:

https://github.com/alexisrolland/flask-graphene-sqlalchemy/wiki/Flask-Graphene-SQLAlchemy-Tutorial https://docs.graphene-python.org/projects/sqlalchemy/en/latest/tutorial/

resolve_address​

def resolve_address(obj, _)

Convert 'address' from bytes to string.

Arguments:

  • obj - Object containing address attribute
  • _ - Unused GraphQL parameter

Returns:

  • str - Decoded address string or empty string

resolve_cdpcachedeviceid​

def resolve_cdpcachedeviceid(obj, _)

Convert 'cdpcachedeviceid' from bytes to string.

Arguments:

  • obj - Object containing cdpcachedeviceid attribute
  • _ - Unused GraphQL parameter

Returns:

  • str - Decoded cdpcachedeviceid string or empty string

resolve_cdpcacheplatform​

def resolve_cdpcacheplatform(obj, _)

Convert 'cdpcacheplatform' from bytes to string.

Arguments:

  • obj - Object containing cdpcacheplatform attribute
  • _ - Unused GraphQL parameter

Returns:

  • str - Decoded cdpcacheplatform string or empty string

resolve_cdpcachedeviceport​

def resolve_cdpcachedeviceport(obj, _)

Convert 'cdpcachedeviceport' from bytes to string.

Arguments:

  • obj - Object containing cdpcachedeviceport attribute
  • _ - Unused GraphQL parameter

Returns:

  • str - Decoded cdpcachedeviceport string or empty string

resolve_hostname​

def resolve_hostname(obj, _)

Convert 'hostname' from bytes to string.

Arguments:

  • obj - Object containing hostname attribute
  • _ - Unused GraphQL parameter

Returns:

  • str - Decoded hostname string or empty string

resolve_ifalias​

def resolve_ifalias(obj, _)

Convert 'ifalias' from bytes to string.

Arguments:

  • obj - Object containing ifalias attribute
  • _ - Unused GraphQL parameter

Returns:

  • str - Decoded ifalias string or empty string

resolve_ifname​

def resolve_ifname(obj, _)

Convert 'ifname' from bytes to string.

Arguments:

  • obj - Object containing ifname attribute
  • _ - Unused GraphQL parameter

Returns:

  • str - Decoded ifname string or empty string

resolve_ifdescr​

def resolve_ifdescr(obj, _)

Convert 'ifdescr' from bytes to string.

Arguments:

  • obj - Object containing ifdescr attribute
  • _ - Unused GraphQL parameter

Returns:

  • str - Decoded ifdescr string or empty string

resolve_lldpremportdesc​

def resolve_lldpremportdesc(obj, _)

Convert 'lldpremportdesc' from bytes to string.

Arguments:

  • obj - Object containing lldpremportdesc attribute
  • _ - Unused GraphQL parameter

Returns:

  • str - Decoded lldpremportdesc string or empty string

resolve_lldpremsyscapenabled​

def resolve_lldpremsyscapenabled(obj, _)

Convert 'lldpremsyscapenabled' from bytes to string.

Arguments:

  • obj - Object containing lldpremsyscapenabled attribute
  • _ - Unused GraphQL parameter

Returns:

  • str - Decoded lldpremsyscapenabled string or empty string

resolve_lldpremsysdesc​

def resolve_lldpremsysdesc(obj, _)

Convert 'lldpremsysdesc' from bytes to string.

Arguments:

  • obj - Object containing lldpremsysdesc attribute
  • _ - Unused GraphQL parameter

Returns:

  • str - Decoded lldpremsysdesc string or empty string

resolve_lldpremsysname​

def resolve_lldpremsysname(obj, _)

Convert 'lldpremsysname' from bytes to string.

Arguments:

  • obj - Object containing lldpremsysname attribute
  • _ - Unused GraphQL parameter

Returns:

  • str - Decoded lldpremsysname string or empty string

resolve_mac​

def resolve_mac(obj, _)

Convert 'mac' from bytes to string.

Arguments:

  • obj - Object containing mac attribute
  • _ - Unused GraphQL parameter

Returns:

  • str - Decoded mac string or empty string

resolve_organization​

def resolve_organization(obj, _)

Convert 'organization' from bytes to string.

Arguments:

  • obj - Object containing organization attribute
  • _ - Unused GraphQL parameter

Returns:

  • str - Decoded organization string or empty string

resolve_device_by_hostname​

def resolve_device_by_hostname(self, info, hostname=None, **kwargs)

Resolve devices by hostname for historical data.

If no hostname is provided, returns all enabled devices.

Arguments:

  • info Any - GraphQL info context.
  • hostname str, optional - The hostname of the device to filter by. Defaults to None.
  • **kwargs - Additional keyword arguments.

Returns:

  • sqlalchemy.orm.query.Query - Query object with the matching devices.

resolve_name​

def resolve_name(obj, _)

Convert 'name' from bytes to string.

Arguments:

  • obj - Object containing name attribute
  • _ - Unused GraphQL parameter

Returns:

  • str - Decoded name string or empty string

resolve_notes​

def resolve_notes(obj, _)

Convert 'notes' from bytes to string.

Arguments:

  • obj - Object containing notes attribute
  • _ - Unused GraphQL parameter

Returns:

  • str - Decoded notes string or empty string

resolve_oui​

def resolve_oui(obj, _)

Convert 'oui' from bytes to string.

Arguments:

  • obj - Object containing oui attribute
  • _ - Unused GraphQL parameter

Returns:

  • str - Decoded oui string or empty string

resolve_sys_description​

def resolve_sys_description(obj, _)

Convert 'sys_description' from bytes to string.

Arguments:

  • obj - Object containing sys_description attribute
  • _ - Unused GraphQL parameter

Returns:

  • str - Decoded sys_description string or empty string

resolve_sys_name​

def resolve_sys_name(obj, _)

Convert 'sys_name' from bytes to string.

Arguments:

  • obj - Object containing sys_name attribute
  • _ - Unused GraphQL parameter

Returns:

  • str - Decoded sys_name string or empty string

resolve_sys_uptime​

def resolve_sys_uptime(obj, _)

Convert 'sys_uptime' from Null to zero.

Arguments:

  • obj - Object containing sys_uptime attribute
  • _ - Unused GraphQL parameter

Returns:

  • float - System uptime value or 0 if null

resolve_sys_objectid​

def resolve_sys_objectid(obj, _)

Convert 'sys_objectid' from bytes to string.

Arguments:

  • obj - Object containing sys_objectid attribute
  • _ - Unused GraphQL parameter

Returns:

  • str - Decoded sys_objectid string or empty string

EventAttribute Objects​

class EventAttribute()

Descriptive attributes of the Event table.

A generic class to mutualize description of attributes for both queries and mutations.

RootAttribute Objects​

class RootAttribute()

Descriptive attributes of the Event table.

A generic class to mutualize description of attributes for both queries and mutations.

ZoneAttribute Objects​

class ZoneAttribute()

Descriptive attributes of the Zone table.

A generic class to mutualize description of attributes for both queries and mutations.

OuiAttribute Objects​

class OuiAttribute()

Descriptive attributes of the Oui table.

A generic class to mutualize description of attributes for both queries and mutations.

DeviceAttribute Objects​

class DeviceAttribute()

Descriptive attributes of the Device table.

A generic class to mutualize description of attributes for both queries and mutations.

SystemStatAttribute Objects​

class SystemStatAttribute()

Descriptive attributes of the SystemStat table.

A generic class to mutualize description of attributes for both queries and mutations.

L1InterfaceAttribute Objects​

class L1InterfaceAttribute()

Descriptive attributes of the L1Interface table.

A generic class to mutualize description of attributes for both queries and mutations.

VlanAttribute Objects​

class VlanAttribute()

Descriptive attributes of the Vlan table.

A generic class to mutualize description of attributes for both queries and mutations.

VlanPortAttribute Objects​

class VlanPortAttribute()

Descriptive attributes of the VlanPort table.

A generic class to mutualize description of attributes for both queries and mutations.

MacAttribute Objects​

class MacAttribute()

Descriptive attributes of the Mac table.

A generic class to mutualize description of attributes for both queries and mutations.

MacIpAttribute Objects​

class MacIpAttribute()

Descriptive attributes of the MacIp table.

A generic class to mutualize description of attributes for both queries and mutations.

MacPortAttribute Objects​

class MacPortAttribute()

Descriptive attributes of the MacPort table.

A generic class to mutualize description of attributes for both queries and mutations.

IpAttribute Objects​

class IpAttribute()

Descriptive attributes of the MacPort table.

A generic class to mutualize description of attributes for both queries and mutations.

IpPortAttribute Objects​

class IpPortAttribute()

Descriptive attributes of the MacPort table.

A generic class to mutualize description of attributes for both queries and mutations.

api

Module of switchmap API routes.

Contains all routes that the Flask API uses

api.routes.config

Config API routes for Switchmap.

merge_preserving_secrets​

def merge_preserving_secrets(current, incoming)

Merge two configuration objects while preserving secret values.

Arguments:

  • current dict | Any - Existing configuration or value.
  • incoming dict | Any - New configuration or value to merge.

Returns:

  • result - Merged configuration where secrets are preserved.

read_config​

def read_config()

Read the configuration file from disk.

Arguments:

None

Returns:

  • dict - The loaded configuration data. Returns an empty dictionary if the configuration file does not exist or is empty.

write_config​

def write_config(data)

Write the configuration data to disk.

Arguments:

  • data dict - The configuration data to write.

Returns:

None

get_config​

@API_CONFIG.route("/config", methods=["GET"])
def get_config()

Return the current configuration as JSON.

Arguments:

None

Returns:

  • Response - A Flask JSON response containing the current config loaded from config.yaml.

mask_secrets​

def mask_secrets(config: dict) -> dict

Recursively masks sensitive values in a configuration dictionary.

Specifically, replaces the value of "db_pass" with a masked string, while preserving the structure of nested dictionaries.

Arguments:

  • config dict - The configuration dictionary to process.

Returns:

  • dict - A new dictionary with secrets masked.

post_config​

@API_CONFIG.route("/config", methods=["POST"])
def post_config()

Update the config.yaml with new JSON data from the request.

Arguments:

None

Returns:

  • Response - A Flask JSON response indicating success or failure. Returns 400 if the JSON data is invalid, otherwise returns a success message.

deep_merge​

def deep_merge(dst, src)

Recursively merge two dictionaries or values.

Arguments:

  • dst dict | Any - Destination dictionary or value.
  • src dict | Any - Source dictionary or value to merge into dst.

Returns:

  • result - Result of merging src into dst.

patch_config​

@API_CONFIG.route("/config", methods=["PATCH"])
def patch_config()

Partially update the SwitchMap configuration.

Handles the db_pass secret:

  • Expects a dictionary with a key "new".
  • Updates db_pass directly.
  • Other non-secret fields are merged directly.

Arguments:

None

The request JSON body can contain:

  • "db_pass" (dict, optional): a dictionary with a key "new" containing the new password
  • Other configuration keys to update.

Returns:

  • Response - JSON response indicating success or failure:
    • 400 if the request JSON is invalid or db_pass format is incorrect.
    • 200 with a JSON response: status set to "success"

api.routes.post

Database server API. HTTP POST routes.

post_device_data​

@API_POST.route(API_POLLER_POST_URI, methods=["POST"])
def post_device_data()

Accept posts of network device data from pollers.

Arguments:

None

Returns:

  • _response - OK message when successful

post_searchterm​

@API_POST.route(API_POLLER_SEARCH_URI, methods=["POST"])
def post_searchterm()

Accept posts searches.

Arguments:

None

Returns:

  • _response - OK message when successful

api.routes.graphql

GraphQL routes.

api.routes

Define the switchmap.server.api.routes package.

Arguments:

None

Returns:

None