Add AmqpDsn class

changes/3254-kludex.md

@@ -0,0 +1 @@
+Add RabbitMqDsn class

docs/examples/settings_main.py

@@ -6,6 +6,7 @@
     PyObject,
     RedisDsn,
     PostgresDsn,
+    RabbitMqDsn,
     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'
+    rabbitmq_dsn: RabbitMqDsn = 'amqp://guest:guest@rabbitmq:5672//'
 
     special_function: PyObject = 'math.cos'
 

docs/usage/types.md

@@ -492,6 +492,9 @@ _(This script is complete, it should run "as is")_
 `PostgresDsn`
 : a postgres DSN style URL; see [URLs](#urls)
 
+`RabbitMqDsn`
+: a RabbitMQ DSN style URL; see [URLs](#urls)
+
 `RedisDsn`
 : a redis DSN style URL; see [URLs](#urls)
 
@@ -584,6 +587,7 @@ For URI/URL validation the following types are available:
   - `postgresql+psycopg2cffi`
   - `postgresql+py-postgresql`
   - `postgresql+pygresql`
+- `RabbitMqDsn`: schema `amqp`, user info required, TLD 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',
+    'RabbitMqDsn',
     'RedisDsn',
     'KafkaDsn',
     'validate_email',

pydantic/networks.py

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

tests/test_networks.py

@@ -9,6 +9,7 @@
     KafkaDsn,
     NameEmail,
     PostgresDsn,
+    RabbitMqDsn,
     RedisDsn,
     ValidationError,
     stricturl,
@@ -443,6 +444,24 @@ class Model(BaseModel):
     assert error == {'loc': ('a',), 'msg': 'URL host invalid', 'type': 'value_error.url.host'}
 
 
+def test_rabbitmq_dsns():
+    class Model(BaseModel):
+        a: RabbitMqDsn
+
+    assert Model(a='amqp://user:pass@localhost:5432/app').a == 'amqp://user:pass@localhost:5432/app'
+    assert Model(a='amqp://user:pass@localhost:5432//').a == 'amqp://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'
+    assert exc_info.value.json().startswith('[')
+
+    with pytest.raises(ValidationError) as exc_info:
+        Model(a='amqp://localhost:5432/app')
+    error = exc_info.value.errors()[0]
+    assert error == {'loc': ('a',), 'msg': 'userinfo required in URL but missing', 'type': 'value_error.url.userinfo'}
+
+
 def test_redis_dsns():
     class Model(BaseModel):
         a: RedisDsn