Ftp and Websocket connection strings support

pydantic/__init__.py

@@ -109,6 +109,9 @@
     'AnyHttpUrl',
     'FileUrl',
     'HttpUrl',
+    'FtpUrl',
+    'WebsocketUrl',
+    'AnyWebsocketUrl',
     'UrlConstraints',
     'EmailStr',
     'NameEmail',
@@ -263,6 +266,9 @@
     'AnyHttpUrl': (__package__, '.networks'),
     'FileUrl': (__package__, '.networks'),
     'HttpUrl': (__package__, '.networks'),
+    'FtpUrl': (__package__, '.networks'),
+    'WebsocketUrl': (__package__, '.networks'),
+    'AnyWebsocketUrl': (__package__, '.networks'),
     'UrlConstraints': (__package__, '.networks'),
     'EmailStr': (__package__, '.networks'),
     'NameEmail': (__package__, '.networks'),

pydantic/networks.py

@@ -28,7 +28,10 @@
     'AnyUrl',
     'AnyHttpUrl',
     'FileUrl',
+    'FtpUrl',
     'HttpUrl',
+    'WebsocketUrl',
+    'AnyWebsocketUrl',
     'UrlConstraints',
     'EmailStr',
     'NameEmail',
@@ -184,11 +187,30 @@ class MyModel(BaseModel):
     Also, Chrome, Firefox, and Safari all currently accept `http://exam_ple.com` as a URL, so we're in good
     (or at least big) company.
 """
+AnyWebsocketUrl = Annotated[Url, UrlConstraints(allowed_schemes=['ws', 'wss'])]
+"""A type that will accept any ws or wss URL.
+
+* TLD not required
+* Host required
+"""
+WebsocketUrl = Annotated[Url, UrlConstraints(max_length=2083, allowed_schemes=['ws', 'wss'])]
+"""A type that will accept any ws or wss URL.
+
+* TLD required
+* Host required
+* Max length 2083
+"""
 FileUrl = Annotated[Url, UrlConstraints(allowed_schemes=['file'])]
 """A type that will accept any file URL.
 
 * Host not required
 """
+FtpUrl = Annotated[Url, UrlConstraints(allowed_schemes=['ftp'])]
+"""A type that will accept ftp URL.
+
+* TLD not required
+* Host required
+"""
 PostgresDsn = Annotated[
     MultiHostUrl,
     UrlConstraints(

tests/test_networks.py

@@ -12,6 +12,7 @@
     ClickHouseDsn,
     CockroachDsn,
     FileUrl,
+    FtpUrl,
     HttpUrl,
     KafkaDsn,
     MariaDBDsn,
@@ -24,6 +25,7 @@
     Strict,
     UrlConstraints,
     ValidationError,
+    WebsocketUrl,
 )
 from pydantic.networks import validate_email
 
@@ -368,6 +370,104 @@ class Model(BaseModel):
     assert str(m.v) == expected_str
 
 
+@pytest.mark.parametrize(
+    'value,expected',
+    [
+        ('ws://example.com', 'ws://example.com/'),
+        ('wss://example.com', 'wss://example.com/'),
+        ('wss://ws.example.com/', 'wss://ws.example.com/'),
+        ('ws://ws.example.com/', 'ws://ws.example.com/'),
+        ('ws://example.com:8080', 'ws://example.com:8080/'),
+        ('ws://example.com/path', 'ws://example.com/path'),
+        ('wss://example.com:4433', 'wss://example.com:4433/'),
+        ('wss://example.com/path', 'wss://example.com/path'),
+    ],
+)
+def test_websocket_url_success(value, expected):
+    class Schema(BaseModel):
+        ws: WebsocketUrl
+
+    assert Schema(ws=value).ws.unicode_string() == expected
+
+
+@pytest.mark.parametrize(
+    'value,expected',
+    [
+        ('ws://example.com', 80),
+        ('wss://example.com', 443),
+        ('wss://ws.example.com/', 443),
+        ('ws://ws.example.com/', 80),
+        ('ws://example.com:8080', 8080),
+        ('ws://example.com:9999/path', 9999),
+        ('wss://example.com:4433', 4433),
+        ('wss://example.com/path', 443),
+    ],
+)
+def test_websocket_url_port_success(value, expected):
+    class Schema(BaseModel):
+        ws: WebsocketUrl
+
+    assert Schema(ws=value).ws.port == expected
+
+
+@pytest.mark.parametrize(
+    'value,expected',
+    [
+        ('ws://example.com', '/'),
+        ('wss://example.com', '/'),
+        ('wss://ws.example.com/', '/'),
+        ('ws://ws.example.com/', '/'),
+        ('ws://example.com:8080', '/'),
+        ('ws://example.com:9999/path', '/path'),
+        ('wss://example.com:4433', '/'),
+        ('wss://example.com/path/to/ws', '/path/to/ws'),
+    ],
+)
+def test_websocket_url_path_success(value, expected):
+    class Schema(BaseModel):
+        ws: WebsocketUrl
+
+    assert Schema(ws=value).ws.path == expected
+
+
+@pytest.mark.parametrize(
+    'value,expected',
+    [
+        ('ftp://example.com', 'ftp://example.com/'),
+        ('ftp://example.com/path/to/ftp', 'ftp://example.com/path/to/ftp'),
+        ('ftp://example.com:21', 'ftp://example.com/'),
+        ('ftp://example.com:21/path/to/ftp', 'ftp://example.com/path/to/ftp'),
+        ('ftp://example.com', 'ftp://example.com/'),
+        ('ftp://example.com/path/to/ftp', 'ftp://example.com/path/to/ftp'),
+        ('ftp://example.com:990', 'ftp://example.com:990/'),
+        ('ftp://example.com:990/path/to/ftp', 'ftp://example.com:990/path/to/ftp'),
+    ],
+)
+def test_ftp_url_success(value, expected):
+    class Schema(BaseModel):
+        ftp: FtpUrl
+
+    assert Schema(ftp=value).ftp.unicode_string() == expected
+
+
+@pytest.mark.parametrize(
+    'value,expected',
+    [
+        ('ftp://example.com', 21),
+        ('ftp://example.com/path/to/ftp', 21),
+        ('ftp://example.com:21', 21),
+        ('ftp://exaMplФ.com:221/path/to/ftp', 221),
+        ('ftp://example.com:144', 144),
+        ('ftp://example.com:990/path/to/ftp', 990),
+    ],
+)
+def test_ftp_url_port_success(value, expected):
+    class Schema(BaseModel):
+        ftp: FtpUrl
+
+    assert Schema(ftp=value).ftp.port == expected
+
+
 @pytest.mark.parametrize(
     'dsn',
     [