Skip to content

Utility models and protocols

ClientProvider

Bases: Protocol

Client provider protocol for FastAPI database dependencies.

Source code in fastapi_motor_oil/model.py 
24
25
26
27
28
29
30
class ClientProvider(Protocol):
    """
    Client provider protocol for FastAPI database dependencies.
    """

    def __call__(self) -> AgnosticClient:
        ...

DatabaseProvider

Bases: Protocol

Database provider protocol for FastAPI database dependencies.

Source code in fastapi_motor_oil/model.py 
33
34
35
36
37
38
39
class DatabaseProvider(Protocol):
    """
    Database provider protocol for FastAPI database dependencies.
    """

    def __call__(self) -> AgnosticDatabase:
        ...

DeleteResultModel

Bases: BaseModel

Delete result model.

Source code in fastapi_motor_oil/model.py 
109
110
111
112
113
114
class DeleteResultModel(BaseModel):
    """
    Delete result model.
    """

    delete_count: int

DocumentModel

Bases: BaseModel

Pydantic base model for MongoDB documents.

It exposes the _id attribute as id.

Source code in fastapi_motor_oil/model.py 
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
class DocumentModel(BaseModel):
    """
    Pydantic base model for MongoDB documents.

    It exposes the `_id` attribute as `id`.
    """

    id: StrObjectId = Field(alias="_id")

    class Config:
        arbitrary_types_allowed = True
        json_encoders = {ObjectId: str}

StrObjectId

Bases: ObjectId

Custom BSON ObjectId for use with Pydantic.

Source code in fastapi_motor_oil/model.py 
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
class StrObjectId(ObjectId):
    """
    Custom BSON `ObjectId` for use with Pydantic.
    """

    @classmethod
    def __get_validators__(cls) -> Generator[Callable[[Any], StrObjectId], None, None]:
        yield cls.validate

    @classmethod
    def validate(cls, value: Any) -> StrObjectId:
        """
        Checks whether the given value is a valid `ObjectId`"""
        if not ObjectId.is_valid(value):
            raise ValueError("Invalid StrObjectId")
        return cls(value)

    @classmethod
    def __modify_schema__(cls, field_schema: dict[str, Any]) -> None:
        field_schema.update(type="string")

validate(value) classmethod

Checks whether the given value is a valid ObjectId

Source code in fastapi_motor_oil/model.py 
82
83
84
85
86
87
88
@classmethod
def validate(cls, value: Any) -> StrObjectId:
    """
    Checks whether the given value is a valid `ObjectId`"""
    if not ObjectId.is_valid(value):
        raise ValueError("Invalid StrObjectId")
    return cls(value)

UTCDatetime

Bases: datetime

Pydantic datetime field that enforces UTC timezone.

Source code in fastapi_motor_oil/model.py 
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
class UTCDatetime(datetime):
    """
    Pydantic datetime field that enforces UTC timezone.
    """

    @classmethod
    def __get_validators__(cls) -> Generator[Any, None, None]:
        yield parse_datetime  # default pydantic behavior
        yield cls.ensure_utc

    @classmethod
    def ensure_utc(cls, value: datetime) -> datetime:
        """
        Makes sure the given datetime is in UTC.

        If `value` has no timezone info, the method sets UTC.

        Raises:
            ValueError: If `value` has timezone info but it's not UTC.
        """
        tzinfo = value.tzinfo
        if tzinfo is None:  # No timezone info, assume UTC.
            return value.replace(tzinfo=timezone.utc)

        if tzinfo == timezone.utc:  # Timezone is UTC, no-op.
            return value

        # Non-UTC timezone info, raise exception.
        raise ValueError("Non-UTC timezone.")

ensure_utc(value) classmethod

Makes sure the given datetime is in UTC.

If value has no timezone info, the method sets UTC.

Raises:

Type Description
ValueError

If value has timezone info but it's not UTC.
Source code in fastapi_motor_oil/model.py 
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
@classmethod
def ensure_utc(cls, value: datetime) -> datetime:
    """
    Makes sure the given datetime is in UTC.

    If `value` has no timezone info, the method sets UTC.

    Raises:
        ValueError: If `value` has timezone info but it's not UTC.
    """
    tzinfo = value.tzinfo
    if tzinfo is None:  # No timezone info, assume UTC.
        return value.replace(tzinfo=timezone.utc)

    if tzinfo == timezone.utc:  # Timezone is UTC, no-op.
        return value

    # Non-UTC timezone info, raise exception.
    raise ValueError("Non-UTC timezone.")