25
25
import json
26
26
import logging
27
27
import re
28
- from typing import Any , Dict , List , Literal , Optional , Union
28
+ from typing import Any , Dict , List , Literal , Optional , Union , Self
29
29
30
30
# Third-Party
31
31
from pydantic import AnyHttpUrl , BaseModel , ConfigDict , Field , field_serializer , field_validator , model_validator , ValidationInfo
@@ -1518,9 +1518,9 @@ class GatewayRead(BaseModelWithConfigDict):
1518
1518
slug : str = Field (None , description = "Slug for gateway endpoint URL" )
1519
1519
1520
1520
# This will be the main method to automatically populate fields
1521
- @classmethod
1522
1521
@model_validator (mode = "after" )
1523
- def _populate_auth (cls , values : Dict [str , Any ]) -> Dict [str , Any ]:
1522
+ @classmethod
1523
+ def _populate_auth (cls , values : Self ) -> Dict [str , Any ]:
1524
1524
"""Populate authentication fields based on auth_type and encoded auth_value.
1525
1525
1526
1526
This post-validation method decodes the stored authentication value and
@@ -1552,28 +1552,33 @@ def _populate_auth(cls, values: Dict[str, Any]) -> Dict[str, Any]:
1552
1552
1553
1553
Examples:
1554
1554
>>> # Basic auth example
1555
- >>> values = GatewayRead._populate_auth({
1556
- ... 'auth_type': 'basic',
1557
- ... 'auth_value': encode_auth({'username': 'admin', 'password': 'secret'})
1558
- ... })
1555
+ >>> string_bytes = "admin:secret".encode("utf-8")
1556
+ >>> encoded_auth = base64.urlsafe_b64encode(string_bytes).decode("utf-8")
1557
+ >>> values = GatewayRead.model_construct(
1558
+ ... auth_type="basic",
1559
+ ... auth_value=encode_auth({"Authorization": f"Basic {encoded_auth}"})
1560
+ ... )
1561
+ >>> values = GatewayRead._populate_auth(values)
1559
1562
>>> values.auth_username
1560
1563
'admin'
1561
1564
>>> values.auth_password
1562
1565
'secret'
1563
1566
1564
1567
>>> # Bearer auth example
1565
- >>> values = GatewayRead._populate_auth({
1566
- ... 'auth_type': 'bearer',
1567
- ... 'auth_value': encode_auth({'Authorization': 'Bearer mytoken123'})
1568
- ... })
1568
+ >>> values = GatewayRead.model_construct(
1569
+ ... auth_type="bearer",
1570
+ ... auth_value=encode_auth({"Authorization": "Bearer mytoken123"})
1571
+ ... )
1572
+ >>> values = GatewayRead._populate_auth(values)
1569
1573
>>> values.auth_token
1570
1574
'mytoken123'
1571
1575
1572
1576
>>> # Custom headers example
1573
- >>> values = GatewayRead._populate_auth({
1574
- ... 'auth_type': 'authheaders',
1575
- ... 'auth_value': encode_auth({'X-API-Key': 'abc123'})
1576
- ... })
1577
+ >>> values = GatewayRead.model_construct(
1578
+ ... auth_type='authheaders',
1579
+ ... auth_value=encode_auth({"X-API-Key": "abc123"})
1580
+ ... )
1581
+ >>> values = GatewayRead._populate_auth(values)
1577
1582
>>> values.auth_header_key
1578
1583
'X-API-Key'
1579
1584
>>> values.auth_header_value
@@ -1583,8 +1588,9 @@ def _populate_auth(cls, values: Dict[str, Any]) -> Dict[str, Any]:
1583
1588
auth_value_encoded = values .auth_value
1584
1589
auth_value = decode_auth (auth_value_encoded )
1585
1590
if auth_type == "basic" :
1586
- u = auth_value .get ("username" )
1587
- p = auth_value .get ("password" )
1591
+ auth = auth_value .get ("Authorization" )
1592
+ auth = auth .removeprefix ("Basic " )
1593
+ u , p = base64 .urlsafe_b64decode (auth ).decode ("utf-8" ).split (":" )
1588
1594
if not u or not p :
1589
1595
raise ValueError ("basic auth requires both username and password" )
1590
1596
values .auth_username , values .auth_password = u , p
0 commit comments