Add NATS DSN

pydantic/__init__.py

@@ -117,6 +117,7 @@
     'RedisDsn',
     'MongoDsn',
     'KafkaDsn',
+    'NatsDsn',
     'MySQLDsn',
     'MariaDBDsn',
     'validate_email',

pydantic/networks.py

@@ -40,6 +40,7 @@
     'RedisDsn',
     'MongoDsn',
     'KafkaDsn',
+    'NatsDsn',
     'validate_email',
     'MySQLDsn',
     'MariaDBDsn',
@@ -309,6 +310,16 @@ def check_db_name(cls, v):
 * TLD not required
 * Host required
 """
+NatsDsn = Annotated[
+    MultiHostUrl, UrlConstraints(allowed_schemes=['nats', 'tls', 'ws'], default_host='localhost', default_port=4222)
+]
+"""A type that will accept any NATS DSN.
+
+NATS is a connective technology built for the ever increasingly hyper-connected world.
+It is a single technology that enables applications to securely communicate across
+any combination of cloud vendors, on-premise, edge, web and mobile, and devices.
+More: https://nats.io
+"""
 MySQLDsn = Annotated[
     Url,
     UrlConstraints(

tests/test_networks.py

@@ -16,6 +16,7 @@
     MongoDsn,
     MySQLDsn,
     NameEmail,
+    NatsDsn,
     PostgresDsn,
     RedisDsn,
     Strict,
@@ -661,6 +662,23 @@ class Model(BaseModel):
     assert m.a.password is None
 
 
+@pytest.mark.parametrize(
+    'dsn,result',
+    [
+        ('nats://user:pass@localhost:4222', 'nats://user:pass@localhost:4222'),
+        ('tls://user@localhost', 'tls://user@localhost:4222'),
+        ('ws://localhost:2355', 'ws://localhost:2355/'),
+        ('tls://', 'tls://localhost:4222'),
+        ('ws://:password@localhost:9999', 'ws://:password@localhost:9999/'),
+    ],
+)
+def test_nats_dsns(dsn, result):
+    class Model(BaseModel):
+        dsn: NatsDsn
+
+    assert str(Model(dsn=dsn).dsn) == result
+
+
 def test_custom_schemes():
     class Model(BaseModel):
         v: Annotated[Url, UrlConstraints(allowed_schemes=['ws', 'wss']), Strict()]