from pydantic import BaseModel, ConfigDict, GetCoreSchemaHandler
from pydantic_core import core_schema
from typing import Type, Any, Dict  

class House:
    humidity: int
    owner: str

    def __init__(self, value: str):
        self.house_name = value

    @classmethod
    def validate(cls, v=None, *args, **kwargs):
        return House(v)

    @classmethod
    def __get_pydantic_core_schema__(
        cls, source: Type[Any], handler: GetCoreSchemaHandler

    ) -> core_schema.CoreSchema:
        return core_schema.general_plain_validator_function(cls.validate)

def test_serialize():
    class CreditorOverview(BaseModel):
        outputs: Dict[str, House]
        model_config = ConfigDict(
            json_encoders={
                House: lambda v: "*****",
            },
        )

    class FunkoCollection(BaseModel):
        outputs: Dict[str, House]

    creds1 = CreditorOverview(outputs={"secret": House("beach house")})
    # this works because of CreditorOverview.model_config.json_encoder
    print(creds1.model_dump_json())
    # this doesn't work ('Unable to serialize unknown type House')
    creds2 = FunkoCollection(outputs={"secret": House("money shack")})
    print(creds2.model_dump_json())

test_serialize()

        return core_schema.no_info_after_validator_function(
            cls.validate,
            core_schema.str_schema(),
            serialization=core_schema.format_ser_schema("?????"),
        )

rowkey bilbao fucked around with this message at 11:39 on Sep 26, 2023

class Gem:
    def __init__(self, value):
        self.value = value

from pydantic import BaseModel, ConfigDict
from pydantic.functional_serializers import PlainSerializer
from typing_extensions import Annotated

from typing import Type, Any, Dict, Union
class SimpleAsteroid(BaseModel):
    contains: Annotated[
        Gem,
        PlainSerializer(
            lambda x: f"{x.value[::-1]}", return_type=str, when_used="json"
        ),
    ]
    model_config = ConfigDict(arbitrary_types_allowed=True)

m1 = SimpleAsteroid(contains=Gem("sploppa"))
print(m1.model_dump_json())

m2 = ComplicatedAsteroid(soil={"underground": Gem("sploppa")})
m3 = ComplicatedAsteroid(soil={"herb": "excellent"})
m4 = ComplicatedAsteroid(soil="rich")

class ComplicatedAsteroid(BaseModel):
    soil: Union[
        Dict[
            str, # Dict key type
            Union[ # Dict value types
                str,
                Annotated[
                    Gem,
                    PlainSerializer(
                        lambda x: f"{x.value[::-1]}", return_type=str, when_used="json"
                    ),
                ],
            ],
        ],
        str,
        None,
    ]
    model_config = ConfigDict(arbitrary_types_allowed=True)

rowkey bilbao fucked around with this message at 14:10 on Sep 27, 2023

from pydantic import BaseModel, ConfigDict
from pydantic.functional_serializers import PlainSerializer
from typing_extensions import Annotated

class GemBase:
    def __init__(self, value):
        self.value = value

Gem = Annotated[
    GemBase,
    PlainSerializer(lambda x: f"it is {x.value}", return_type=str, when_used="json"),
]

class SimpleAsteroid(BaseModel):
    contains: Gem
    model_config = ConfigDict(arbitrary_types_allowed=True)

class ComplicatedAsteroid(BaseModel):
    soil: Union[
        Dict[
            Any,
            Union[Dict, Gem],
        ], str
    ]
    model_config = ConfigDict(arbitrary_types_allowed=True)
 

m1 = SimpleAsteroid(contains=Gem("sploppa"))
m2 = ComplicatedAsteroid(soil={"underground": Gem("a diamond")})
m3 = ComplicatedAsteroid(soil={"underground": {"rocks": Gem("a spirulin")}})
 

print(m1.model_dump_json()) # OK
print(m2.model_dump_json()) # OK
print(m3.model_dump_json()) # NOK: 'Unable to serialize unknown type Gem'

rowkey bilbao fucked around with this message at 17:08 on Sep 27, 2023

SurgicalOntologist fucked around with this message at 21:20 on Sep 27, 2023

Son of Thunderbeast fucked around with this message at 02:37 on Oct 1, 2023

Jabor fucked around with this message at 02:24 on Oct 1, 2023

def generate_index(id:int, table_name:str):
        s = sql_module.Tools()
        existing_pids = set(s.get_columns(table=table_name,cols=["index"])["index"].tolist())
        p_int = random.randint(1,65000)
        p_salt = p_int.to_bytes(16, "little")
        index = pyblake2.blake2b(data=bytes(id), salt=p_salt)
        while index in existing_pids:
            p_int = random.randint(1,65000)
            p_salt = p_int.to_bytes(16, "little")
            index = pyblake2.blake2b(data=bytes(id), salt=p_salt)
        return index.hexdigest()

Son of Thunderbeast fucked around with this message at 03:32 on Oct 1, 2023