Skip to content

ManyToMany class

saffier.ManyToManyField

ManyToManyField(to, through=None, **kwargs)

Bases: Field

Representation of a ManyToManyField based on a foreignkey.

PARAMETER DESCRIPTION
to

TYPE: Union[Type[Model], str]

through

TYPE: Optional[Type[Model]] DEFAULT: None

**kwargs

TYPE: Any DEFAULT: {}

Source code in saffier/core/db/fields/base.py
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
def __init__(
    self,
    to: typing.Union[typing.Type["Model"], str],
    through: typing.Optional[typing.Type["Model"]] = None,
    **kwargs: typing.Any,
):
    if "null" in kwargs:
        terminal.write_warning("Declaring `null` on a ManyToMany relationship has no effect.")

    super().__init__(null=True)
    self.to = to
    self.through = through
    self.related_name = kwargs.pop("related_name", None)

    if self.related_name:
        assert isinstance(self.related_name, str), "related_name must be a string."

    self.related_name = self.related_name.lower() if self.related_name else None

server_default instance-attribute

server_default = pop('server_default', None)

null instance-attribute

null = get('null', False)

default_value instance-attribute

default_value = get('default', None)

primary_key instance-attribute

primary_key = primary_key

index instance-attribute

index = index

unique instance-attribute

unique = unique

validator instance-attribute

validator = get_validator(**kwargs)

comment instance-attribute

comment = get('comment', None)

owner instance-attribute

owner = pop('owner', None)

server_onupdate instance-attribute

server_onupdate = pop('server_onupdate', None)

autoincrement instance-attribute

autoincrement = pop('autoincrement', False)

secret instance-attribute

secret = pop('secret', False)

to instance-attribute

to = to

through instance-attribute

through = through

related_name instance-attribute

related_name = lower() if related_name else None

target property

target

get_validator

get_validator(**kwargs)
PARAMETER DESCRIPTION
**kwargs

TYPE: Any DEFAULT: {}

Source code in saffier/core/db/fields/base.py
85
86
def get_validator(self, **kwargs: typing.Any) -> SaffierField:
    return SaffierField(**kwargs)  # pragma: no cover

get_column_type

get_column_type()
Source code in saffier/core/db/fields/base.py
88
89
def get_column_type(self) -> sqlalchemy.types.TypeEngine:
    raise NotImplementedError()  # pragma: no cover

get_constraints

get_constraints()
Source code in saffier/core/db/fields/base.py
91
92
def get_constraints(self) -> typing.Any:
    return []

expand_relationship

expand_relationship(value)
PARAMETER DESCRIPTION
value

TYPE: Any

Source code in saffier/core/db/fields/base.py
94
95
def expand_relationship(self, value: typing.Any) -> typing.Any:
    return value

raise_for_non_default

raise_for_non_default(default, server_default)
PARAMETER DESCRIPTION
default

TYPE: Any

server_default

TYPE: Any

Source code in saffier/core/db/fields/base.py
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
def raise_for_non_default(self, default: typing.Any, server_default: typing.Any) -> typing.Any:
    has_default: bool = True
    has_server_default: bool = True

    if default is None or default is False:
        has_default = False
    if server_default is None or server_default is False:
        has_server_default = False

    if (
        not isinstance(self, (IntegerField, BigIntegerField))
        and not has_default
        and not has_server_default
    ):
        raise ValueError(
            "Primary keys other then IntegerField and BigIntegerField, must provide a default or a server_default."
        )

get_column

get_column(name)
PARAMETER DESCRIPTION
name

TYPE: str

Source code in saffier/core/db/fields/base.py
361
362
363
364
365
366
367
368
369
370
371
372
373
374
def get_column(self, name: str) -> sqlalchemy.Column:
    target = self.target
    to_field = target.fields[target.pkname]

    column_type = to_field.get_column_type()
    constraints = [
        sqlalchemy.schema.ForeignKey(
            f"{target.meta.tablename}.{target.pkname}",
            ondelete=saffier.CASCADE,
            onupdate=saffier.CASCADE,
            name=self.get_fk_name(name=name),
        )
    ]
    return sqlalchemy.Column(name, column_type, *constraints, nullable=self.null)

add_model_to_register

add_model_to_register(model)

Adds the model to the registry to make sure it can be generated by the Migrations

PARAMETER DESCRIPTION
model

TYPE: Type[Model]

Source code in saffier/core/db/fields/base.py
376
377
378
379
380
def add_model_to_register(self, model: typing.Type["Model"]) -> None:
    """
    Adds the model to the registry to make sure it can be generated by the Migrations
    """
    self.registry.models[model.__name__] = model

get_fk_name

get_fk_name(name)

Builds the fk name for the engine. Engines have a limitation of the foreign key being bigger than 63 characters. if that happens, we need to assure it is small.

PARAMETER DESCRIPTION
name

TYPE: str

Source code in saffier/core/db/fields/base.py
382
383
384
385
386
387
388
389
390
391
392
def get_fk_name(self, name: str) -> str:
    """
    Builds the fk name for the engine.
    Engines have a limitation of the foreign key being bigger than 63
    characters.
    if that happens, we need to assure it is small.
    """
    fk_name = f"fk_{self.owner.meta.tablename}_{self.target.meta.tablename}_{self.target.pkname}_{name}"
    if not len(fk_name) > CHAR_LIMIT:
        return fk_name
    return fk_name[:CHAR_LIMIT]

create_through_model

create_through_model()

Creates the default empty through model.

Generates a middle model based on the owner of the field and the field itself and adds it to the main registry to make sure it generates the proper models and migrations.

Source code in saffier/core/db/fields/base.py
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
def create_through_model(self) -> typing.Any:
    """
    Creates the default empty through model.

    Generates a middle model based on the owner of the field and the field itself and adds
    it to the main registry to make sure it generates the proper models and migrations.
    """
    self.to = typing.cast(typing.Type["Model"], self.target)

    if self.through:
        if isinstance(self.through, str):
            self.through = self.owner.meta.registry.models[self.through]  # type: ignore

        self.through.meta.is_multi = True
        self.through.meta.multi_related = [self.to.__name__.lower()]
        return self.through

    owner_name = self.owner.__name__
    to_name = self.to.__name__
    class_name = f"{owner_name}{to_name}"
    tablename = f"{owner_name.lower()}s_{to_name}s".lower()

    new_meta_namespace = {
        "tablename": tablename,
        "registry": self.owner.meta.registry,
        "is_multi": True,
        "multi_related": [to_name.lower()],
    }

    new_meta = type("MetaInfo", (), new_meta_namespace)

    # Define the related names
    owner_related_name = (
        f"{self.related_name}_{class_name.lower()}s_set"
        if self.related_name
        else f"{owner_name.lower()}_{class_name.lower()}s_set"
    )

    to_related_name = (
        f"{self.related_name}"
        if self.related_name
        else f"{to_name.lower()}_{class_name.lower()}s_set"
    )

    through_model = type(
        class_name,
        (saffier.Model,),
        {
            "Meta": new_meta,
            "id": saffier.IntegerField(primary_key=True),
            f"{owner_name.lower()}": ForeignKey(
                self.owner,
                on_delete=saffier.CASCADE,
                null=True,
                related_name=owner_related_name,
            ),
            f"{to_name.lower()}": ForeignKey(
                self.to, on_delete=saffier.CASCADE, null=True, related_name=to_related_name
            ),
        },
    )
    self.through = typing.cast(typing.Type["Model"], through_model)

    self.add_model_to_register(self.through)