models.py

        # Force the string to lowercase
        lower_string = string.lower()
        if HOST_NAME_RE.fullmatch(lower_string) is None:
            raise ValidationError("Interface name shall match [a-z0-9\-]{2,20}")
        return lower_string

    def stack_members(self):
        """Return all items part of the stack sorted by stack member number"""
        members = [item for item in self.items if item.stack_member is not None]
        return sorted(members, key=lambda x: x.stack_member)

    def stack_members_numbers(self):
        """Return the list of stack member numbers"""
        return [item.stack_member for item in self.stack_members()]

    def free_stack_members(self):
        """Return the list of free stack member numbers"""
        return [nb for nb in range(0, 10) if nb not in self.stack_members_numbers()]

    def to_dict(self):
        d = super().to_dict()
        d.update(
            {
                "name": self.name,
                "device_type": str(self.device_type),
                "model": self.model,
                "description": self.description,
                "items": [str(item) for item in self.items],
                "interfaces": [str(interface) for interface in self.interfaces],
                "ansible_vars": self.ansible_vars,
                "ansible_groups": [str(group) for group in self.ansible_groups],
            }
        )
        return d


class Interface(CreatedMixin, db.Model):
    network_id = db.Column(db.Integer, db.ForeignKey("network.id"), nullable=False)
    ip = db.Column(postgresql.INET, nullable=False, unique=True)
    name = db.Column(db.Text, nullable=False, unique=True)
    mac_id = db.Column(db.Integer, db.ForeignKey("mac.id"))
    host_id = db.Column(db.Integer, db.ForeignKey("host.id"))

    # Add delete and delete-orphan options to automatically delete cnames when:
    # - deleting an interface
    # - de-associating a cname (removing it from the interface.cnames list)
    cnames = db.relationship(
        "Cname", backref="interface", cascade="all, delete, delete-orphan"
    )
    tags = db.relationship(
        "Tag",
        secondary=interfacetags_table,
        lazy="subquery",
        backref=db.backref("interfaces", lazy=True),
    )

    def __init__(self, **kwargs):
        # Always set self.network and not self.network_id to call validate_interfaces
        network_id = kwargs.pop("network_id", None)
        if network_id is not None:
            kwargs["network"] = Network.query.get(network_id)
        elif "network" in kwargs:
            # Automatically convert network to an instance of Network if it was passed
            # as a string
            kwargs["network"] = utils.convert_to_model(
                kwargs["network"], Network, "vlan_name"
            )
        # WARNING! Setting self.network will call validate_interfaces in the Network class
        # For the validation to work, self.ip must be set before!
        # Ensure that ip is passed before network
        try:
            ip = kwargs.pop("ip")
        except KeyError:
            super().__init__(**kwargs)
        else:
            super().__init__(ip=ip, **kwargs)

    @validates("name")
    def validate_name(self, key, string):
        """Ensure the name matches the required format"""
        if string is None:
            return None
        # Force the string to lowercase
        lower_string = string.lower()
        if HOST_NAME_RE.fullmatch(lower_string) is None:
            raise ValidationError("Interface name shall match [a-z0-9\-]{2,20}")
        return lower_string

    @property
    def address(self):
        return ipaddress.ip_address(self.ip)

    @property
    def is_ioc(self):
        for tag in self.tags:
            if tag.name == "IOC":
                return True
        return False

    def __str__(self):
        return str(self.name)

    def __repr__(self):
        return f"Interface(id={self.id}, network_id={self.network_id}, ip={self.ip}, name={self.name}, mac={self.mac})"

    def to_dict(self):
        d = super().to_dict()
        d.update(
            {
                "network": str(self.network),
                "ip": self.ip,
                "name": self.name,
                "mac": utils.format_field(self.mac),
                "host": utils.format_field(self.host),
                "cnames": [str(cname) for cname in self.cnames],
                "domain": str(self.network.domain),
                "tags": [str(tag) for tag in self.tags],
            }
        )
        if self.host:
            d["device_type"] = str(self.host.device_type)
            d["model"] = utils.format_field(self.host.model)
        else:
            d["device_type"] = None
            d["model"] = None
        return d


class Mac(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    address = db.Column(postgresql.MACADDR, nullable=False, unique=True)
    item_id = db.Column(db.Integer, db.ForeignKey("item.id"))

    interfaces = db.relationship("Interface", backref="mac")

    def __str__(self):
        return str(self.address)

    @validates("address")
    def validate_address(self, key, string):
        """Ensure the address is a valid MAC address"""
        if string is None:
            return None
        if MAC_ADDRESS_RE.fullmatch(string) is None:
            raise ValidationError(f"'{string}' does not appear to be a MAC address")
        return string

    def to_dict(self):
        return {
            "id": self.id,
            "address": self.address,
            "item": utils.format_field(self.item),
            "interfaces": [str(interface) for interface in self.interfaces],
        }


class Cname(CreatedMixin, db.Model):
    name = db.Column(db.Text, nullable=False, unique=True)
    interface_id = db.Column(db.Integer, db.ForeignKey("interface.id"), nullable=False)

    def __str__(self):
        return str(self.name)

    def to_dict(self):
        d = super().to_dict()
        d.update({"name": self.name, "interface": str(self.interface)})
        return d


class Domain(CreatedMixin, db.Model):
    name = db.Column(db.Text, nullable=False, unique=True)

    scopes = db.relationship("NetworkScope", backref="domain")
    networks = db.relationship("Network", backref="domain")

    def __str__(self):
        return str(self.name)

    def to_dict(self):
        d = super().to_dict()
        d.update(
            {
                "name": self.name,
                "scopes": [str(scope) for scope in self.scopes],
                "networks": [str(network) for network in self.networks],
            }
        )
        return d


class NetworkScope(CreatedMixin, db.Model):
    __tablename__ = "network_scope"
    name = db.Column(CIText, nullable=False, unique=True)
    first_vlan = db.Column(db.Integer, nullable=False, unique=True)
    last_vlan = db.Column(db.Integer, nullable=False, unique=True)
    supernet = db.Column(postgresql.CIDR, nullable=False, unique=True)
    domain_id = db.Column(db.Integer, db.ForeignKey("domain.id"), nullable=False)
    description = db.Column(db.Text)

    networks = db.relationship("Network", backref="scope")

    __table_args__ = (
        sa.CheckConstraint(
            "first_vlan < last_vlan", name="first_vlan_less_than_last_vlan"
        ),
    )

    def __str__(self):
        return str(self.name)

    @property
    def supernet_ip(self):
        return ipaddress.ip_network(self.supernet)

    def prefix_range(self):
        """Return the list of subnet prefix that can be used for this network scope"""
        return list(range(self.supernet_ip.prefixlen + 1, 31))

    def vlan_range(self):
        """Return the list of vlan ids that can be assigned for this network scope

        The range is defined by the first and last vlan
        """
        return range(self.first_vlan, self.last_vlan + 1)

    def used_vlans(self):
        """Return the list of vlan ids in use

        The list is sorted
        """
        return sorted(network.vlan_id for network in self.networks)

    def available_vlans(self):
        """Return the list of vlan ids available"""
        return [vlan for vlan in self.vlan_range() if vlan not in self.used_vlans()]

    def used_subnets(self):
        """Return the list of subnets in use

        The list is sorted
        """
        return sorted(network.network_ip for network in self.networks)

    def available_subnets(self, prefix):
        """Return the list of available subnets with the given prefix"""
        return [
            str(subnet)
            for subnet in self.supernet_ip.subnets(new_prefix=prefix)
            if subnet not in self.used_subnets()
        ]

    def to_dict(self):
        d = super().to_dict()
        d.update(
            {
                "name": self.name,
                "first_vlan": self.first_vlan,
                "last_vlan": self.last_vlan,
                "supernet": self.supernet,
                "description": self.description,
                "domain": str(self.domain),
                "networks": [str(network) for network in self.networks],
            }
        )
        return d


# Define RQ JobStatus as a Python enum
# We can't use the one defined in rq/job.py as it's
# not a real enum (it's a custom one) and is not
# compatible with sqlalchemy
class JobStatus(Enum):
    QUEUED = "queued"
    FINISHED = "finished"
    FAILED = "failed"
    STARTED = "started"
    DEFERRED = "deferred"


class Task(db.Model):
    # Use job id generated by RQ
    id = db.Column(postgresql.UUID, primary_key=True)
    created_at = db.Column(db.DateTime, default=utcnow())
    ended_at = db.Column(db.DateTime)
    name = db.Column(db.Text, nullable=False, index=True)
    command = db.Column(db.Text)
    status = db.Column(db.Enum(JobStatus, name="job_status"))
    awx_job_id = db.Column(db.Integer)
    exception = db.Column(db.Text)
    user_id = db.Column(
        db.Integer,
        db.ForeignKey("user_account.id"),
        nullable=False,
        default=utils.fetch_current_user_id,
    )

    @property
    def awx_job_url(self):
        if self.awx_job_id is None:
            return None
        return urllib.parse.urljoin(
            current_app.config["AWX_URL"], f"/#/jobs/{self.awx_job_id}"
        )

    def __str__(self):
        return str(self.id)

    def to_dict(self):
        return {
            "id": self.id,
            "name": self.name,
            "created_at": utils.format_field(self.created_at),
            "ended_at": utils.format_field(self.ended_at),
            "status": self.status.name,
            "awx_job_id": self.awx_job_id,
            "awx_job_url": self.awx_job_url,
            "command": self.command,
            "exception": self.exception,
            "user": str(self.user),
        }


@sa.event.listens_for(db.session, "before_flush")
def before_flush(session, flush_context, instances):
    """Before flush hook

    Used to trigger core services update on any Interface modification.

    See http://docs.sqlalchemy.org/en/latest/orm/session_events.html#before-flush
    """
    for instance in itertools.chain(session.new, session.dirty, session.deleted):
        if isinstance(instance, Interface):
            # In session.dirty, we could check session.is_modified(instance)
            # but it's always True due to the updated_at column.
            utils.trigger_core_services_update()
            break


# call configure_mappers after defining all the models
# required by sqlalchemy_continuum
sa.orm.configure_mappers()
ItemVersion = version_class(Item)