Repository: tophat/ormar-postgres-extensions (language: python)

Issue:
Support network address types
### Is your feature request related to a problem? Please describe.

Postgres has [network address](https://www.postgresql.org/docs/current/datatype-net-types.html) types that should be supported by this library

### Describe the solution you'd like

1. Fields for different network address types
2. Implementation of [network address operators](https://www.postgresql.org/docs/current/functions-net.html)

### Describe alternatives you've considered

_No response_

### Additional context

_No response_


Discussion / hints:

Fix patch:
diff --git a/README.md b/README.md
index 025c242..09eb05a 100644
--- a/README.md
+++ b/README.md
@@ -138,6 +138,88 @@ The maps to the [`overlap`](https://docs.sqlalchemy.org/en/14/dialects/postgresq
 await ModelWithArray.objects.filter(data__array_overlap=["a"]).all()
 ```
 
+
+#### INET / CIDR
+
+```python
+from ipaddress import IPv4Address, IPv6Address, IPv4Interface, IPv6Interface
+from typing import Union
+
+import ormar
+import ormar_postgres_extensions as ormar_pg_ext
+
+IPAddress = Union[
+    IPv4Address,
+    IPv4Interface,
+    IPv6Address,
+    IPv6Interface,
+]
+
+class INETTestModel(ormar.Model):
+    id: int = ormar.Integer(primary_key=True)
+    inet: IPAddress = ormar_pg_ext.INET()
+    cidr: IPAddress = ormar_pg_ext.CIDR()
+```
+
+##### contained_by
+
+This maps to the [`<<` operator](https://www.postgresql.org/docs/current/functions-net.html)
+
+```python
+from ipaddress import ip_interface
+await INETTestModel.objects.filter(inet__contained_by=ip_interface("192.168.1.0/24")).all()
+```
+
+##### contained_by_eq
+
+This maps to the [`<<=` operator](https://www.postgresql.org/docs/current/functions-net.html)
+
+```python
+from ipaddress import ip_interface
+await INETTestModel.objects.filter(inet__contained_by_eq=ip_interface("192.168.1.0/24")).all()
+```
+
+##### contains_subnet
+
+This maps to the [`>>` operator](https://www.postgresql.org/docs/current/functions-net.html)
+
+```python
+from ipaddress import ip_interface
+await INETTestModel.objects.filter(inet__contains_subnet=ip_interface("192.168.1.0/24")).all()
+```
+
+##### contains_subnet_eq
+
+This maps to the [`>>=` operator](https://www.postgresql.org/docs/current/functions-net.html)
+
+```python
+from ipaddress import ip_interface
+await INETTestModel.objects.filter(inet__contains_subnet_eq=ip_interface("192.168.1.0/24")).all()
+```
+
+##### contains_or_eq
+
+This maps to the [`&&` operator](https://www.postgresql.org/docs/current/functions-net.html)
+
+```python
+from ipaddress import ip_interface
+await INETTestModel.objects.filter(inet__contains_or_eq=ip_interface("192.168.1.0/24")).all()
+```
+
+#### MACADDR
+
+```python
+from ipaddress import IPv4Address, IPv6Address, IPv4Interface, IPv6Interface
+from typing import Union
+
+import ormar
+import ormar_postgres_extensions as ormar_pg_ext
+
+class MacAddrTestModel(ormar.Model):
+    id: int = ormar.Integer(primary_key=True)
+    addr: str = ormar_pg_ext.MACADDR()
+```
+
 ## Uninstalling
 
 ```python
diff --git a/src/ormar_postgres_extensions/__init__.py b/src/ormar_postgres_extensions/__init__.py
index f34eab7..8463d21 100644
--- a/src/ormar_postgres_extensions/__init__.py
+++ b/src/ormar_postgres_extensions/__init__.py
@@ -1,5 +1,8 @@
 from .fields import (  # noqa: F401
     ARRAY,
+    CIDR,
+    INET,
     JSONB,
+    MACADDR,
     UUID,
 )
diff --git a/src/ormar_postgres_extensions/fields/__init__.py b/src/ormar_postgres_extensions/fields/__init__.py
index 3e4fceb..d2e4b91 100644
--- a/src/ormar_postgres_extensions/fields/__init__.py
+++ b/src/ormar_postgres_extensions/fields/__init__.py
@@ -1,3 +1,6 @@
 from .array import ARRAY  # noqa: F401
+from .cidr import CIDR  # noqa: F401
+from .inet import INET  # noqa: F401
 from .jsonb import JSONB  # noqa: F401
+from .macaddr import MACADDR  # noqa: F401
 from .uuid import UUID  # noqa: F401
diff --git a/src/ormar_postgres_extensions/fields/cidr.py b/src/ormar_postgres_extensions/fields/cidr.py
new file mode 100644
index 0000000..aaf2905
--- /dev/null
+++ b/src/ormar_postgres_extensions/fields/cidr.py
@@ -0,0 +1,51 @@
+from ipaddress import (
+    IPv4Address,
+    IPv4Interface,
+    IPv6Address,
+    IPv6Interface,
+)
+from typing import (
+    Any,
+    Union,
+)
+
+import ormar
+from sqlalchemy.dialects import postgresql
+from sqlalchemy.sql.expression import Operators
+from sqlalchemy.types import TypeDecorator
+
+
+class PostgresCidrTypeDecorator(TypeDecorator):
+    """
+    Postgres specific CIDR type for user with Ormar
+
+    A custom TypeDecorator is needed here to be able to access the operators
+    that are not provided by SQLAlchemy directly as functions. These are implemented
+    inside of the comparator_factory class
+    """
+
+    impl = postgresql.CIDR
+
+    class comparator_factory(postgresql.INET.Comparator):
+        def contained_by(self, value):
+            return Operators.op(self, "<<")(value)
+
+        def contained_by_eq(self, value):
+            return Operators.op(self, "<<=")(value)
+
+        def contains_subnet(self, value):
+            return Operators.op(self, ">>")(value)
+
+        def contains_subnet_eq(self, value):
+            return Operators.op(self, ">>=")(value)
+
+        def contains_or_eq(self, value):
+            return Operators.op(self, "&&")(value)
+
+
+class CIDR(ormar.fields.model_fields.ModelFieldFactory, str):
+    _type = Union[IPv4Address, IPv6Address, IPv4Interface, IPv6Interface]
+
+    @classmethod
+    def get_column_type(cls, **kwargs: Any) -> postgresql.CIDR:
+        return PostgresCidrTypeDecorator()
diff --git a/src/ormar_postgres_extensions/fields/inet.py b/src/ormar_postgres_extensions/fields/inet.py
new file mode 100644
index 0000000..259993a
--- /dev/null
+++ b/src/ormar_postgres_extensions/fields/inet.py
@@ -0,0 +1,139 @@
+from ipaddress import (
+    IPv4Address,
+    IPv4Interface,
+    IPv6Address,
+    IPv6Interface,
+)
+from typing import (
+    Any,
+    Union,
+)
+
+import ormar
+from sqlalchemy.dialects import postgresql
+from sqlalchemy.sql.expression import Operators
+from sqlalchemy.types import TypeDecorator
+
+
+def contained_by(self, other: Any) -> ormar.queryset.clause.FilterGroup:
+    """
+    works as postgresql `column << inet VALUE`
+    :param other: value to check against operator
+    :type other: Any
+    :return: FilterGroup for operator
+    :rtype: ormar.queryset.clause.FilterGroup
+    """
+    return self._select_operator(op="contained_by", other=other)
+
+
+def contained_by_eq(self, other: Any) -> ormar.queryset.clause.FilterGroup:
+    """
+    works as postgresql `column <<= inet VALUE`
+    :param other: value to check against operator
+    :type other: Any
+    :return: FilterGroup for operator
+    :rtype: ormar.queryset.clause.FilterGroup
+    """
+    return self._select_operator(op="contained_by_eq", other=other)
+
+
+def contains_subnet(self, other: Any) -> ormar.queryset.clause.FilterGroup:
+    """
+    works as postgresql `column >> inet VALUE`
+    :param other: value to check against operator
+    :type other: Any
+    :return: FilterGroup for operator
+    :rtype: ormar.queryset.clause.FilterGroup
+    """
+    return self._select_operator(op="contains", other=other)
+
+
+def contains_subnet_eq(self, other: Any) -> ormar.queryset.clause.FilterGroup:
+    """
+    works as postgresql `column >>= inet VALUE`
+    :param other: value to check against operator
+    :type other: Any
+    :return: FilterGroup for operator
+    :rtype: ormar.queryset.clause.FilterGroup
+    """
+    return self._select_operator(op="contains_subnet_eq", other=other)
+
+
+def contains_or_eq(self, other: Any) -> ormar.queryset.clause.FilterGroup:
+    """
+    works as postgresql `column >>= inet VALUE`
+    :param other: value to check against operator
+    :type other: Any
+    :return: FilterGroup for operator
+    :rtype: ormar.queryset.clause.FilterGroup
+    """
+    return self._select_operator(op="contains_or_eq", other=other)
+
+
+# Need to patch the filter objects to support JSONB specifc actions
+FIELD_ACCESSOR_MAP = [
+    ("contained_by", contained_by),
+    ("contained_by_eq", contained_by_eq),
+    ("contains_subnet", contains_subnet),
+    ("contains_subnet_eq", contains_subnet_eq),
+    ("contains_or_eq", contains_or_eq),
+]
+
+
+for (method_name, method) in FIELD_ACCESSOR_MAP:
+    setattr(ormar.queryset.FieldAccessor, method_name, method)
+
+
+# These lines allow Ormar to lookup the new filter methods and map
+# it to the correct PGSQL functions
+ACCESSOR_MAP = [
+    ("contained_by", "contained_by"),
+    ("contained_by_eq", "contained_by_eq"),
+    ("contains_subnet", "contains_subnet"),
+    ("contains_subnet_eq", "contains_subnet_eq"),
+    ("contains_or_eq", "contains_or_eq"),
+]
+
+for (ormar_operation, pg_operation) in ACCESSOR_MAP:
+    ormar.queryset.actions.filter_action.FILTER_OPERATORS[
+        ormar_operation
+    ] = pg_operation
+    ormar.queryset.actions.filter_action.METHODS_TO_OPERATORS[
+        ormar_operation
+    ] = ormar_operation
+
+
+class PostgresInetTypeDecorator(TypeDecorator):
+    """
+    Postgres specific INET type for user with Ormar
+
+    A custom TypeDecorator is needed here to be able to access the operators
+    that are not provided by SQLAlchemy directly as functions. These are implemented
+    inside of the comparator_factory class
+    """
+
+    impl = postgresql.INET
+
+    class comparator_factory(postgresql.INET.Comparator):
+        def contained_by(self, value):
+            return Operators.op(self, "<<")(value)
+
+        def contained_by_eq(self, value):
+            return Operators.op(self, "<<=")(value)
+
+        def contains_subnet(self, value):
+            return Operators.op(self, ">>")(value)
+
+        def contains_subnet_eq(self, value):
+            return Operators.op(self, ">>=")(value)
+
+        def contains_or_eq(self, value):
+            return Operators.op(self, "&&")(value)
+
+
+class INET(ormar.fields.model_fields.ModelFieldFactory, str):
+    _type = Union[IPv4Address, IPv6Address, IPv4Interface, IPv6Interface]
+
+    @classmethod
+    def get_column_type(cls, **kwargs: Any) -> postgresql.INET:
+        return PostgresInetTypeDecorator()
diff --git a/src/ormar_postgres_extensions/fields/macaddr.py b/src/ormar_postgres_extensions/fields/macaddr.py
new file mode 100644
index 0000000..f6dc2d6
--- /dev/null
+++ b/src/ormar_postgres_extensions/fields/macaddr.py
@@ -0,0 +1,13 @@
+from typing import Any
+
+import ormar
+from sqlalchemy.dialects import postgresql
+
+
+class MACADDR(ormar.fields.model_fields.ModelFieldFactory, str):
+    _type = str
+
+    @classmethod
+    def get_column_type(cls, **kwargs: Any) -> postgresql.MACADDR:
+        # Tell Ormar that this column should be a postgres macaddr type
+        return postgresql.MACADDR()


Test patch:
diff --git a/tests/fields/test_cidr.py b/tests/fields/test_cidr.py
new file mode 100644
index 0000000..eed5a1d
--- /dev/null
+++ b/tests/fields/test_cidr.py
@@ -0,0 +1,223 @@
+from ipaddress import (
+    IPv4Address,
+    IPv4Interface,
+    IPv6Address,
+    IPv6Interface,
+    ip_address,
+    ip_interface,
+)
+from typing import (
+    Optional,
+    Union,
+)
+
+import ormar
+import pytest
+
+import ormar_postgres_extensions as ormar_pg_ext
+from tests.database import (
+    database,
+    metadata,
+)
+
+IPAddress = Union[
+    IPv4Address,
+    IPv4Interface,
+    IPv6Address,
+    IPv6Interface,
+]
+
+
+class CidrTestModel(ormar.Model):
+    class Meta:
+        database = database
+        metadata = metadata
+
+    id: int = ormar.Integer(primary_key=True)
+    cidr: IPAddress = ormar_pg_ext.CIDR()
+
+
+class NullableCidrTestModel(ormar.Model):
+    class Meta:
+        database = database
+        metadata = metadata
+
+    id: int = ormar.Integer(primary_key=True)
+    cidr: Optional[IPAddress] = ormar_pg_ext.CIDR(nullable=True)
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize(
+    "ip, ip_type",
+    [
+        # IP V4
+        ("192.168.0.1/32", IPv4Interface),
+        # IP V6
+        ("::ffff:c0a8:1/128", IPv6Interface),
+    ],
+)
+async def test_create_model_with_cidr_address_specified(db, ip, ip_type):
+    created = await CidrTestModel(cidr=ip_interface(ip)).save()
+    assert str(created.cidr) == ip
+    assert isinstance(created.cidr, ip_type)
+
+    # Confirm the model got saved to the DB by querying it back
+    found = await CidrTestModel.objects.get()
+    assert found.cidr == created.cidr
+    assert isinstance(found.cidr, ip_type)
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize(
+    "ip, ip_type",
+    [
+        # IP V4 Subnet
+        ("192.168.1.0/24", IPv4Interface),
+        # IP V6 Subnet
+        ("2001:db8::/120", IPv6Interface),
+    ],
+)
+async def test_create_model_with_cidr_network_specified(db, ip, ip_type):
+    created = await CidrTestModel(cidr=ip_interface(ip)).save()
+    assert str(created.cidr) == ip
+    assert isinstance(created.cidr, ip_type)
+
+    # Confirm the model got saved to the DB by querying it back
+    found = await CidrTestModel.objects.get()
+    assert found.cidr == created.cidr
+    assert isinstance(found.cidr, ip_type)
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize(
+    "ip",
+    [
+        "192.168.0.1",
+        "::ffff:c0a8:1",
+    ],
+)
+async def test_get_model_by_cidr(db, ip):
+    created = await CidrTestModel(cidr=ip_address(ip)).save()
+
+    found = await CidrTestModel.objects.filter(cidr=ip_address(ip)).all()
+    assert len(found) == 1
+    assert found[0] == created
+
+
+@pytest.mark.asyncio
+async def test_create_model_with_nullable_cidr(db):
+    created = await NullableCidrTestModel().save()
+    assert created.cidr is None
+
+
+@pytest.mark.asyncio
+async def test_get_model_with_nullable_cidr(db):
+    created = await NullableCidrTestModel().save()
+
+    # Ensure querying a model with a null UUID works
+    found = await NullableCidrTestModel.objects.get()
+    assert found == created
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize(
+    "ip, contained, expected",
+    [
+        (ip_address("192.168.1.5"), ip_interface("192.168.1.0/24"), True),
+        (ip_address("192.168.0.5"), ip_interface("192.168.1.0/24"), False),
+        (ip_interface("192.168.1.0/24"), ip_interface("192.168.1.0/24"), False),
+    ],
+)
+async def test_cidr_contained_by(db, ip, contained, expected):
+    created = await CidrTestModel(cidr=ip).save()
+
+    found = await CidrTestModel.objects.filter(cidr__contained_by=contained).all()
+
+    if expected:
+        assert len(found) == 1
+        assert found[0] == created
+    else:
+        assert len(found) == 0
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize(
+    "ip, contained, expected",
+    [
+        (ip_address("192.168.1.5"), ip_interface("192.168.1.0/24"), True),
+        (ip_address("192.168.0.5"), ip_interface("192.168.1.0/24"), False),
+        (ip_interface("192.168.1.0/24"), ip_interface("192.168.1.0/24"), True),
+    ],
+)
+async def test_cidr_contained_by_eq(db, ip, contained, expected):
+    created = await CidrTestModel(cidr=ip).save()
+
+    found = await CidrTestModel.objects.filter(cidr__contained_by_eq=contained).all()
+
+    if expected:
+        assert len(found) == 1
+        assert found[0] == created
+    else:
+        assert len(found) == 0
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize(
+    "ip, contained, expected",
+    [
+        (ip_interface("192.168.1.0/24"), ip_address("192.168.1.5"), True),
+        (ip_interface("192.168.1.0/24"), ip_address("192.168.2.5"), False),
+        (ip_interface("192.168.1.0/24"), ip_interface("192.168.1.0/24"), False),
+    ],
+)
+async def test_contains_subnet(db, ip, contained, expected):
+    created = await CidrTestModel(cidr=ip).save()
+
+    found = await CidrTestModel.objects.filter(cidr__contains_subnet=contained).all()
+
+    if expected:
+        assert len(found) == 1
+        assert found[0] == created
+    else:
+        assert len(found) == 0
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize(
+    "ip, contained, expected",
+    [
+        (ip_interface("192.168.1.0/24"), ip_address("192.168.1.5"), True),
+        (ip_interface("192.168.1.0/24"), ip_address("192.168.2.5"), False),
+        (ip_interface("192.168.1.0/24"), ip_interface("192.168.1.0/24"), True),
+    ],
+)
+async def test_cidr_contains_subnet_eq(db, ip, contained, expected):
+    created = await CidrTestModel(cidr=ip).save()
+
+    found = await CidrTestModel.objects.filter(cidr__contains_subnet_eq=contained).all()
+
+    if expected:
+        assert len(found) == 1
+        assert found[0] == created
+    else:
+        assert len(found) == 0
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize(
+    "ip, contained, expected",
+    [
+        (ip_interface("192.168.1.0/24"), ip_interface("192.168.1.80/28"), True),
+        (ip_interface("192.168.1.0/24"), ip_interface("192.168.2.0/28"), False),
+    ],
+)
+async def test_cidr_contains_or_eq(db, ip, contained, expected):
+    created = await CidrTestModel(cidr=ip).save()
+
+    found = await CidrTestModel.objects.filter(cidr__contains_or_eq=contained).all()
+
+    if expected:
+        assert len(found) == 1
+        assert found[0] == created
+    else:
+        assert len(found) == 0
diff --git a/tests/fields/test_inet.py b/tests/fields/test_inet.py
new file mode 100644
index 0000000..27c6651
--- /dev/null
+++ b/tests/fields/test_inet.py
@@ -0,0 +1,223 @@
+from ipaddress import (
+    IPv4Address,
+    IPv4Interface,
+    IPv6Address,
+    IPv6Interface,
+    ip_address,
+    ip_interface,
+)
+from typing import (
+    Optional,
+    Union,
+)
+
+import ormar
+import pytest
+
+import ormar_postgres_extensions as ormar_pg_ext
+from tests.database import (
+    database,
+    metadata,
+)
+
+IPAddress = Union[
+    IPv4Address,
+    IPv4Interface,
+    IPv6Address,
+    IPv6Interface,
+]
+
+
+class InetTestModel(ormar.Model):
+    class Meta:
+        database = database
+        metadata = metadata
+
+    id: int = ormar.Integer(primary_key=True)
+    inet: IPAddress = ormar_pg_ext.INET()
+
+
+class NullableInetTestModel(ormar.Model):
+    class Meta:
+        database = database
+        metadata = metadata
+
+    id: int = ormar.Integer(primary_key=True)
+    inet: Optional[IPAddress] = ormar_pg_ext.INET(nullable=True)
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize(
+    "ip, ip_type",
+    [
+        # IP V4
+        ("192.168.0.1", IPv4Address),
+        # IP V6
+        ("::ffff:c0a8:1", IPv6Address),
+    ],
+)
+async def test_create_model_with_inet_address_specified(db, ip, ip_type):
+    created = await InetTestModel(inet=ip_address(ip)).save()
+    assert str(created.inet) == ip
+    assert isinstance(created.inet, ip_type)
+
+    # Confirm the model got saved to the DB by querying it back
+    found = await InetTestModel.objects.get()
+    assert found.inet == created.inet
+    assert isinstance(found.inet, ip_type)
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize(
+    "ip, ip_type",
+    [
+        # IP V4 Subnet
+        ("192.168.1.0/24", IPv4Interface),
+        # IP V6 Subnet
+        ("2001:db8::/120", IPv6Interface),
+    ],
+)
+async def test_create_model_with_inet_network_specified(db, ip, ip_type):
+    created = await InetTestModel(inet=ip_interface(ip)).save()
+    assert str(created.inet) == ip
+    assert isinstance(created.inet, ip_type)
+
+    # Confirm the model got saved to the DB by querying it back
+    found = await InetTestModel.objects.get()
+    assert found.inet == created.inet
+    assert isinstance(found.inet, ip_type)
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize(
+    "ip",
+    [
+        "192.168.0.1",
+        "::ffff:c0a8:1",
+    ],
+)
+async def test_get_model_by_inet(db, ip):
+    created = await InetTestModel(inet=ip_address(ip)).save()
+
+    found = await InetTestModel.objects.filter(inet=ip_address(ip)).all()
+    assert len(found) == 1
+    assert found[0] == created
+
+
+@pytest.mark.asyncio
+async def test_create_model_with_nullable_inet(db):
+    created = await NullableInetTestModel().save()
+    assert created.inet is None
+
+
+@pytest.mark.asyncio
+async def test_get_model_with_nullable_inet(db):
+    created = await NullableInetTestModel().save()
+
+    # Ensure querying a model with a null UUID works
+    found = await NullableInetTestModel.objects.get()
+    assert found == created
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize(
+    "ip, contained, expected",
+    [
+        (ip_address("192.168.1.5"), ip_interface("192.168.1.0/24"), True),
+        (ip_address("192.168.0.5"), ip_interface("192.168.1.0/24"), False),
+        (ip_interface("192.168.1.0/24"), ip_interface("192.168.1.0/24"), False),
+    ],
+)
+async def test_inet_contained_by(db, ip, contained, expected):
+    created = await InetTestModel(inet=ip).save()
+
+    found = await InetTestModel.objects.filter(inet__contained_by=contained).all()
+
+    if expected:
+        assert len(found) == 1
+        assert found[0] == created
+    else:
+        assert len(found) == 0
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize(
+    "ip, contained, expected",
+    [
+        (ip_address("192.168.1.5"), ip_interface("192.168.1.0/24"), True),
+        (ip_address("192.168.0.5"), ip_interface("192.168.1.0/24"), False),
+        (ip_interface("192.168.1.0/24"), ip_interface("192.168.1.0/24"), True),
+    ],
+)
+async def test_inet_contained_by_eq(db, ip, contained, expected):
+    created = await InetTestModel(inet=ip).save()
+
+    found = await InetTestModel.objects.filter(inet__contained_by_eq=contained).all()
+
+    if expected:
+        assert len(found) == 1
+        assert found[0] == created
+    else:
+        assert len(found) == 0
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize(
+    "ip, contained, expected",
+    [
+        (ip_interface("192.168.1.0/24"), ip_address("192.168.1.5"), True),
+        (ip_interface("192.168.1.0/24"), ip_address("192.168.2.5"), False),
+        (ip_interface("192.168.1.0/24"), ip_interface("192.168.1.0/24"), False),
+    ],
+)
+async def test_contains_subnet(db, ip, contained, expected):
+    created = await InetTestModel(inet=ip).save()
+
+    found = await InetTestModel.objects.filter(inet__contains_subnet=contained).all()
+
+    if expected:
+        assert len(found) == 1
+        assert found[0] == created
+    else:
+        assert len(found) == 0
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize(
+    "ip, contained, expected",
+    [
+        (ip_interface("192.168.1.0/24"), ip_address("192.168.1.5"), True),
+        (ip_interface("192.168.1.0/24"), ip_address("192.168.2.5"), False),
+        (ip_interface("192.168.1.0/24"), ip_interface("192.168.1.0/24"), True),
+    ],
+)
+async def test_inet_contains_subnet_eq(db, ip, contained, expected):
+    created = await InetTestModel(inet=ip).save()
+
+    found = await InetTestModel.objects.filter(inet__contains_subnet_eq=contained).all()
+
+    if expected:
+        assert len(found) == 1
+        assert found[0] == created
+    else:
+        assert len(found) == 0
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize(
+    "ip, contained, expected",
+    [
+        (ip_interface("192.168.1.0/24"), ip_interface("192.168.1.80/28"), True),
+        (ip_interface("192.168.1.0/24"), ip_interface("192.168.2.0/28"), False),
+    ],
+)
+async def test_inet_contains_or_eq(db, ip, contained, expected):
+    created = await InetTestModel(inet=ip).save()
+
+    found = await InetTestModel.objects.filter(inet__contains_or_eq=contained).all()
+
+    if expected:
+        assert len(found) == 1
+        assert found[0] == created
+    else:
+        assert len(found) == 0
diff --git a/tests/fields/test_macaddr.py b/tests/fields/test_macaddr.py
new file mode 100644
index 0000000..af35ae0
--- /dev/null
+++ b/tests/fields/test_macaddr.py
@@ -0,0 +1,64 @@
+from typing import Optional
+
+import ormar
+import pytest
+
+import ormar_postgres_extensions as ormar_pg_ext
+from tests.database import (
+    database,
+    metadata,
+)
+
+
+class MacAddrTestModel(ormar.Model):
+    class Meta:
+        database = database
+        metadata = metadata
+
+    id: int = ormar.Integer(primary_key=True)
+    addr: str = ormar_pg_ext.MACADDR()
+
+
+class NullableMacAddrTestModel(ormar.Model):
+    class Meta:
+        database = database
+        metadata = metadata
+
+    id: int = ormar.Integer(primary_key=True)
+    addr: Optional[str] = ormar_pg_ext.MACADDR(nullable=True)
+
+
+@pytest.mark.asyncio
+async def test_create_model_with_macaddr_specified(db):
+    created = await MacAddrTestModel(addr="08:00:2b:01:02:03").save()
+    assert str(created.addr) == "08:00:2b:01:02:03"
+    assert isinstance(created.addr, str)
+
+    # Confirm the model got saved to the DB by querying it back
+    found = await MacAddrTestModel.objects.get()
+    assert found.addr == created.addr
+    assert isinstance(found.addr, str)
+
+
+@pytest.mark.asyncio
+async def test_get_model_by_macaddr(db):
+    created = await MacAddrTestModel(addr="08:00:2b:01:02:03").save()
+
+    found = await MacAddrTestModel.objects.filter(addr="08:00:2b:01:02:03").all()
+    assert len(found) == 1
+    assert found[0] == created
+
+
+@pytest.mark.asyncio
+async def test_create_model_with_nullable_macaddr(db):
+    created = await NullableMacAddrTestModel().save()
+    assert created.addr is None
+
+
+@pytest.mark.asyncio
+async def test_get_model_with_nullable_macaddr(db):
+    created = await NullableMacAddrTestModel().save()
+
+    # Ensure querying a model with a null UUID works
+    found = await NullableMacAddrTestModel.objects.get()
+    assert found == created