Add MongoDB network data source name (DSN) schema (#3230)

* Add MongoDsn to pydantic.networks with allowed_schemas and get_default_parts

* Add unit test to MongoDsn and remove default host from default parts becouse it's required by MongoDB protocol

* Fix import issues, follow contributing guide

* Add changes to docs

* Add changes to changelog

changes/3229-snosratiershad.md

@@ -0,0 +1 @@
+Add MongoDB network data source name (DSN) schema

docs/usage/types.md

@@ -549,6 +549,12 @@ _(This script is complete, it should run "as is")_
 `RedisDsn`
 : a redis DSN style URL; see [URLs](#urls)
 
+`MongoDsn`
+: a MongoDB DSN style URL; see [URLs](#urls)
+
+`KafkaDsn`
+: a kafka DSN style URL; see [URLs](#urls)
+
 `stricturl`
 : a type method for arbitrary URL constraints; see [URLs](#urls)
 
@@ -644,6 +650,7 @@ For URI/URL validation the following types are available:
   - `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
+- `MongoDsn` : scheme `mongodb`, user info not required, database name not required, port not required  
   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:
     - `strip_whitespace: bool = True`

pydantic/__init__.py

@@ -59,6 +59,7 @@ __all__ = [
     'PostgresDsn',
     'AmqpDsn',
     'RedisDsn',
+    'MongoDsn',
     'KafkaDsn',
     'validate_email',
     # parse

pydantic/networks.py

@@ -74,6 +74,7 @@ __all__ = [
     'PostgresDsn',
     'AmqpDsn',
     'RedisDsn',
+    'MongoDsn',
     'KafkaDsn',
     'validate_email',
 ]
@@ -394,6 +395,17 @@ class RedisDsn(AnyUrl):
         }
 
 
+class MongoDsn(AnyUrl):
+    allowed_schemes = {'mongodb'}
+
+    # TODO: Needed to generic "Parts" for "Replica Set", "Sharded Cluster", and other mongodb deployment modes
+    @staticmethod
+    def get_default_parts(parts: 'Parts') -> 'Parts':
+        return {
+            'port': '27017',
+        }
+
+
 class KafkaDsn(AnyUrl):
     allowed_schemes = {'kafka'}
 

tests/test_networks.py

@@ -9,6 +9,7 @@ from pydantic import (
     FileUrl,
     HttpUrl,
     KafkaDsn,
+    MongoDsn,
     NameEmail,
     PostgresDsn,
     RedisDsn,
@@ -510,6 +511,33 @@ def test_redis_dsns():
     assert m.a.path == '/0'
 
 
+def test_mongodb_dsns():
+    class Model(BaseModel):
+        a: MongoDsn
+
+    # TODO: Need to unit tests about "Replica Set", "Sharded cluster" and other deployment modes of MongoDB
+    m = Model(a='mongodb://user:pass@localhost:1234/app')
+    assert m.a == 'mongodb://user:pass@localhost:1234/app'
+    assert m.a.user == 'user'
+    assert m.a.password == 'pass'
+
+    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 MongoDB protocol
+    m = Model(a='mongodb://localhost:1234/app')
+    assert m.a == 'mongodb://localhost:1234/app'
+    assert m.a.user is None
+    assert m.a.password is None
+
+    # Only schema and host is required for MongoDB protocol
+    m = Model(a='mongodb://localhost')
+    assert m.a.scheme == 'mongodb'
+    assert m.a.host == 'localhost'
+    assert m.a.port == '27017'
+
+
 def test_kafka_dsns():
     class Model(BaseModel):
         a: KafkaDsn