Add AmqpDsn class

changes/3254-kludex.md

@@ -0,0 +1 @@
+Add `AmqpDsn` class

docs/examples/settings_main.py

@@ -6,6 +6,7 @@
     PyObject,
     RedisDsn,
     PostgresDsn,
+    AmqpDsn,
     Field,
 )
 
@@ -21,6 +22,7 @@ class Settings(BaseSettings):
 
     redis_dsn: RedisDsn = 'redis://user:pass@localhost:6379/1'
     pg_dsn: PostgresDsn = 'postgres://user:pass@localhost:5432/foobar'
+    amqp_dsn: AmqpDsn = 'amqp://user:pass@localhost:5672/'
 
     special_function: PyObject = 'math.cos'
 

docs/usage/types.md

@@ -504,6 +504,9 @@ _(This script is complete, it should run "as is")_
 `PostgresDsn`
 : a postgres DSN style URL; see [URLs](#urls)
 
+`RabbitMqDsn`
+: an `AMQP` DSN style URL as used by RabbitMQ, StormMQ, ActiveMQ etc.; see [URLs](#urls)
+
 `RedisDsn`
 : a redis DSN style URL; see [URLs](#urls)
 
@@ -600,6 +603,7 @@ For URI/URL validation the following types are available:
   - `postgresql+psycopg2cffi`
   - `postgresql+py-postgresql`
   - `postgresql+pygresql`
+- `AmqpDsn`: schema `amqp` or `amqps`, user info not required, TLD not required, host not required
 - `RedisDsn`: scheme `redis` or `rediss`, user info not required, tld not required, host not required (CHANGED: user info
   not required from **v1.6** onwards), user info may be passed without user part (e.g., `rediss://:pass@localhost`)
 - `stricturl`: method with the following keyword arguments:

pydantic/__init__.py

@@ -57,6 +57,7 @@
     'IPvAnyInterface',
     'IPvAnyNetwork',
     'PostgresDsn',
+    'AmqpDsn',
     'RedisDsn',
     'KafkaDsn',
     'validate_email',

pydantic/networks.py

@@ -68,6 +68,7 @@ class Parts(TypedDict, total=False):
     'IPvAnyInterface',
     'IPvAnyNetwork',
     'PostgresDsn',
+    'AmqpDsn',
     'RedisDsn',
     'KafkaDsn',
     'validate_email',
@@ -351,6 +352,11 @@ class PostgresDsn(AnyUrl):
     user_required = True
 
 
+class AmqpDsn(AnyUrl):
+    allowed_schemes = {'amqp', 'amqps'}
+    host_required = False
+
+
 class RedisDsn(AnyUrl):
     allowed_schemes = {'redis', 'rediss'}
     host_required = False

tests/test_networks.py

@@ -1,6 +1,7 @@
 import pytest
 
 from pydantic import (
+    AmqpDsn,
     AnyUrl,
     BaseModel,
     EmailStr,
@@ -443,6 +444,37 @@ class Model(BaseModel):
     assert error == {'loc': ('a',), 'msg': 'URL host invalid', 'type': 'value_error.url.host'}
 
 
+def test_amqp_dsns():
+    class Model(BaseModel):
+        a: AmqpDsn
+
+    m = Model(a='amqp://user:pass@localhost:1234/app')
+    assert m.a == 'amqp://user:pass@localhost:1234/app'
+    assert m.a.user == 'user'
+    assert m.a.password == 'pass'
+
+    m = Model(a='amqps://user:pass@localhost:5432//')
+    assert m.a == 'amqps://user:pass@localhost:5432//'
+
+    with pytest.raises(ValidationError) as exc_info:
+        Model(a='http://example.org')
+    assert exc_info.value.errors()[0]['type'] == 'value_error.url.scheme'
+
+    # Password is not required for AMQP protocol
+    m = Model(a='amqp://localhost:1234/app')
+    assert m.a == 'amqp://localhost:1234/app'
+    assert m.a.user is None
+    assert m.a.password is None
+
+    # Only schema is required for AMQP protocol.
+    # https://www.rabbitmq.com/uri-spec.html
+    m = Model(a='amqps://')
+    assert m.a.scheme == 'amqps'
+    assert m.a.host is None
+    assert m.a.port is None
+    assert m.a.path is None
+
+
 def test_redis_dsns():
     class Model(BaseModel):
         a: RedisDsn