Add rediss (Redis over SSL) protocol to RedisDsn

changes/1911-TrDex.md

@@ -0,0 +1,2 @@
+Add `rediss` (Redis over SSL) protocol to `RedisDsn`
+Allow URLs without `user` part (e.g., `rediss://:pass@localhost`)

docs/examples/types_urls.py

@@ -7,6 +7,7 @@ class MyModel(BaseModel):
 
 m = MyModel(url='http://www.example.com')
 print(m.url)
+
 try:
     MyModel(url='ftp://invalid.url')
 except ValidationError as e:

docs/usage/types.md

@@ -526,8 +526,8 @@ For URI/URL validation the following types are available:
 - `AnyHttpUrl`: schema `http` or `https`, TLD not required
 - `HttpUrl`: schema `http` or `https`, TLD required, max length 2083
 - `PostgresDsn`: schema `postgres` or `postgresql`, user info required, TLD not required
-- `RedisDsn`: schema `redis`, user info not required, tld not required (CHANGED: user info not required from
-  **v1.6** onwards)
+- `RedisDsn`: schema `redis` or `rediss`, user info not required, tld 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:
     - `strip_whitespace: bool = True`
     - `min_length: int = 1`

pydantic/networks.py

@@ -43,7 +43,6 @@
     'validate_email',
 ]
 
-
 _url_regex_cache = None
 _ascii_domain_regex_cache = None
 _int_domain_regex_cache = None
@@ -54,7 +53,7 @@ def url_regex() -> Pattern[str]:
     if _url_regex_cache is None:
         _url_regex_cache = re.compile(
             r'(?:(?P<scheme>[a-z][a-z0-9+\-.]+)://)?'  # scheme https://tools.ietf.org/html/rfc3986#appendix-A
-            r'(?:(?P<user>[^\s:/]+)(?::(?P<password>[^\s/]*))?@)?'  # user info
+            r'(?:(?P<user>[^\s:/]*)(?::(?P<password>[^\s/]*))?@)?'  # user info
             r'(?:'
             r'(?P<ipv4>(?:\d{1,3}\.){3}\d{1,3})|'  # ipv4
             r'(?P<ipv6>\[[A-F0-9]*:[A-F0-9:]+\])|'  # ipv6
@@ -147,8 +146,9 @@ def build(
         url = scheme + '://'
         if user:
             url += user
-            if password:
-                url += ':' + password
+        if password:
+            url += ':' + password
+        if user or password:
             url += '@'
         url += host
         if port:
@@ -183,20 +183,7 @@ def validate(cls, value: Any, field: 'ModelField', config: 'BaseConfig') -> 'Any
         assert m, 'URL regex failed unexpectedly'
 
         parts = m.groupdict()
-        scheme = parts['scheme']
-        if scheme is None:
-            raise errors.UrlSchemeError()
-
-        if cls.allowed_schemes and scheme.lower() not in cls.allowed_schemes:
-            raise errors.UrlSchemePermittedError(cls.allowed_schemes)
-
-        port = parts['port']
-        if port is not None and int(port) > 65_535:
-            raise errors.UrlPortError()
-
-        user = parts['user']
-        if cls.user_required and user is None:
-            raise errors.UrlUserInfoError()
+        parts = cls.validate_parts(parts)
 
         host, tld, host_type, rebuild = cls.validate_host(parts)
 
@@ -205,18 +192,41 @@ def validate(cls, value: Any, field: 'ModelField', config: 'BaseConfig') -> 'Any
 
         return cls(
             None if rebuild else url,
-            scheme=scheme,
-            user=user,
+            scheme=parts['scheme'],
+            user=parts['user'],
             password=parts['password'],
             host=host,
             tld=tld,
             host_type=host_type,
-            port=port,
+            port=parts['port'],
             path=parts['path'],
             query=parts['query'],
             fragment=parts['fragment'],
         )
 
+    @classmethod
+    def validate_parts(cls, parts: Dict[str, str]) -> Dict[str, str]:
+        """
+        A method used to validate parts of an URL.
+        Could be overridden to set default values for parts if missing
+        """
+        scheme = parts['scheme']
+        if scheme is None:
+            raise errors.UrlSchemeError()
+
+        if cls.allowed_schemes and scheme.lower() not in cls.allowed_schemes:
+            raise errors.UrlSchemePermittedError(cls.allowed_schemes)
+
+        port = parts['port']
+        if port is not None and int(port) > 65_535:
+            raise errors.UrlPortError()
+
+        user = parts['user']
+        if cls.user_required and user is None:
+            raise errors.UrlUserInfoError()
+
+        return parts
+
     @classmethod
     def validate_host(cls, parts: Dict[str, str]) -> Tuple[str, Optional[str], str, bool]:
         host, tld, host_type, rebuild = None, None, None, False
@@ -279,7 +289,19 @@ class PostgresDsn(AnyUrl):
 
 
 class RedisDsn(AnyUrl):
-    allowed_schemes = {'redis'}
+    allowed_schemes = {'redis', 'rediss'}
+
+    @classmethod
+    def validate_parts(cls, parts: Dict[str, str]) -> Dict[str, str]:
+        defaults = {
+            'domain': 'localhost' if not (parts['ipv4'] or parts['ipv6']) else '',
+            'port': '6379',
+            'path': '/0',
+        }
+        for key, value in defaults.items():
+            if not parts[key]:
+                parts[key] = value
+        return super().validate_parts(parts)
 
 
 def stricturl(

tests/test_networks.py

@@ -190,6 +190,13 @@ def test_user_no_password():
     assert url.host == 'example.org'
 
 
+def test_user_info_no_user():
+    url = validate_url('http://:password@example.org')
+    assert url.user == ''
+    assert url.password == 'password'
+    assert url.host == 'example.org'
+
+
 def test_at_in_path():
     url = validate_url('https://twitter.com/@handle')
     assert url.scheme == 'https'
@@ -321,21 +328,35 @@ def test_redis_dsns():
     class Model(BaseModel):
         a: RedisDsn
 
-    m = Model(a='redis://user:pass@localhost:5432/app')
-    assert m.a == 'redis://user:pass@localhost:5432/app'
+    m = Model(a='redis://user:pass@localhost:6379/app')
+    assert m.a == 'redis://user:pass@localhost:6379/app'
     assert m.a.user == 'user'
     assert m.a.password == 'pass'
 
+    m = Model(a='rediss://user:pass@localhost:6379/app')
+    assert m.a == 'rediss://user:pass@localhost:6379/app'
+
+    m = Model(a='rediss://:pass@localhost:6379')
+    assert m.a == 'rediss://:pass@localhost:6379/0'
+
     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 redis
-    m = Model(a='redis://localhost:5432/app')
-    assert m.a == 'redis://localhost:5432/app'
+    # Password is not required for Redis protocol
+    m = Model(a='redis://localhost:6379/app')
+    assert m.a == 'redis://localhost:6379/app'
     assert m.a.user is None
     assert m.a.password is None
 
+    # Only schema is required for Redis protocol. Otherwise it will be set to default
+    # https://www.iana.org/assignments/uri-schemes/prov/redis
+    m = Model(a='rediss://')
+    assert m.a.scheme == 'rediss'
+    assert m.a.host == 'localhost'
+    assert m.a.port == '6379'
+    assert m.a.path == '/0'
+
 
 def test_custom_schemes():
     class Model(BaseModel):