add multi column non unique columns
This commit is contained in:
@ -51,6 +51,7 @@ from ormar.fields import (
|
||||
Float,
|
||||
ForeignKey,
|
||||
ForeignKeyField,
|
||||
IndexColumns,
|
||||
Integer,
|
||||
JSON,
|
||||
LargeBinary,
|
||||
@ -102,6 +103,7 @@ __all__ = [
|
||||
"Undefined",
|
||||
"UUID",
|
||||
"UniqueColumns",
|
||||
"IndexColumns",
|
||||
"QuerySetProtocol",
|
||||
"RelationProtocol",
|
||||
"ModelMeta",
|
||||
|
||||
@ -5,7 +5,8 @@ as well as relation Fields (ForeignKey, ManyToMany).
|
||||
Also a definition for custom CHAR based sqlalchemy UUID field
|
||||
"""
|
||||
from ormar.fields.base import BaseField
|
||||
from ormar.fields.foreign_key import ForeignKey, ForeignKeyField, UniqueColumns
|
||||
from ormar.fields.constraints import IndexColumns, UniqueColumns
|
||||
from ormar.fields.foreign_key import ForeignKey, ForeignKeyField
|
||||
from ormar.fields.many_to_many import ManyToMany, ManyToManyField
|
||||
from ormar.fields.model_fields import (
|
||||
BigInteger,
|
||||
@ -36,6 +37,7 @@ __all__ = [
|
||||
"DateTime",
|
||||
"String",
|
||||
"JSON",
|
||||
"IndexColumns",
|
||||
"Integer",
|
||||
"Text",
|
||||
"Float",
|
||||
@ -45,7 +47,6 @@ __all__ = [
|
||||
"ManyToMany",
|
||||
"ManyToManyField",
|
||||
"BaseField",
|
||||
"UniqueColumns",
|
||||
"ForeignKeyField",
|
||||
"ThroughField",
|
||||
"Through",
|
||||
@ -54,4 +55,5 @@ __all__ = [
|
||||
"DECODERS_MAP",
|
||||
"ENCODERS_MAP",
|
||||
"LargeBinary",
|
||||
"UniqueColumns",
|
||||
]
|
||||
|
||||
22
ormar/fields/constraints.py
Normal file
22
ormar/fields/constraints.py
Normal file
@ -0,0 +1,22 @@
|
||||
from typing import Any
|
||||
|
||||
from sqlalchemy import Index, UniqueConstraint
|
||||
|
||||
|
||||
class UniqueColumns(UniqueConstraint):
|
||||
"""
|
||||
Subclass of sqlalchemy.UniqueConstraint.
|
||||
Used to avoid importing anything from sqlalchemy by user.
|
||||
"""
|
||||
|
||||
|
||||
class IndexColumns(Index):
|
||||
def __init__(self, *args: Any, name: str = None) -> None:
|
||||
if not name:
|
||||
name = "TEMPORARY_NAME"
|
||||
super().__init__(name, *args)
|
||||
|
||||
"""
|
||||
Subclass of sqlalchemy.Index.
|
||||
Used to avoid importing anything from sqlalchemy by user.
|
||||
"""
|
||||
@ -18,7 +18,6 @@ from typing import (
|
||||
import sqlalchemy
|
||||
from pydantic import BaseModel, create_model
|
||||
from pydantic.typing import ForwardRef, evaluate_forwardref
|
||||
from sqlalchemy import UniqueConstraint
|
||||
|
||||
import ormar # noqa I101
|
||||
from ormar.exceptions import ModelDefinitionError, RelationshipInstanceError
|
||||
@ -160,13 +159,6 @@ def validate_not_allowed_fields(kwargs: Dict) -> None:
|
||||
)
|
||||
|
||||
|
||||
class UniqueColumns(UniqueConstraint):
|
||||
"""
|
||||
Subclass of sqlalchemy.UniqueConstraint.
|
||||
Used to avoid importing anything from sqlalchemy by user.
|
||||
"""
|
||||
|
||||
|
||||
@dataclass
|
||||
class ForeignKeyConstraint:
|
||||
"""
|
||||
|
||||
@ -285,24 +285,40 @@ def populate_meta_sqlalchemy_table_if_required(meta: "ModelMeta") -> None:
|
||||
|
||||
:param meta: Meta class of the Model without sqlalchemy table constructed
|
||||
:type meta: Model class Meta
|
||||
:return: class with populated Meta.table
|
||||
:rtype: Model class
|
||||
"""
|
||||
if not hasattr(meta, "table") and check_for_null_type_columns_from_forward_refs(
|
||||
meta
|
||||
):
|
||||
for constraint in meta.constraints:
|
||||
if isinstance(constraint, sqlalchemy.UniqueConstraint):
|
||||
constraint.name = (
|
||||
f"uc_{meta.tablename}_"
|
||||
f'{"_".join([str(col) for col in constraint._pending_colargs])}'
|
||||
)
|
||||
set_constraint_names(meta=meta)
|
||||
table = sqlalchemy.Table(
|
||||
meta.tablename, meta.metadata, *meta.columns, *meta.constraints
|
||||
)
|
||||
meta.table = table
|
||||
|
||||
|
||||
def set_constraint_names(meta: "ModelMeta") -> None:
|
||||
"""
|
||||
Populates the names on IndexColumn and UniqueColumns constraints.
|
||||
|
||||
:param meta: Meta class of the Model without sqlalchemy table constructed
|
||||
:type meta: Model class Meta
|
||||
"""
|
||||
for constraint in meta.constraints:
|
||||
if isinstance(constraint, sqlalchemy.UniqueConstraint) and not constraint.name:
|
||||
constraint.name = (
|
||||
f"uc_{meta.tablename}_"
|
||||
f'{"_".join([str(col) for col in constraint._pending_colargs])}'
|
||||
)
|
||||
elif (
|
||||
isinstance(constraint, sqlalchemy.Index)
|
||||
and constraint.name == "TEMPORARY_NAME"
|
||||
):
|
||||
constraint.name = (
|
||||
f"ix_{meta.tablename}_"
|
||||
f'{"_".join([col for col in constraint._pending_colargs])}'
|
||||
)
|
||||
|
||||
|
||||
def update_column_definition(
|
||||
model: Union[Type["Model"], Type["NewBaseModel"]], field: "ForeignKeyField"
|
||||
) -> None:
|
||||
|
||||
@ -17,6 +17,7 @@ import sqlalchemy
|
||||
from sqlalchemy.sql.schema import ColumnCollectionConstraint
|
||||
|
||||
import ormar # noqa I100
|
||||
import ormar.fields.constraints
|
||||
from ormar import ModelDefinitionError # noqa I100
|
||||
from ormar.exceptions import ModelError
|
||||
from ormar.fields import BaseField
|
||||
@ -219,7 +220,8 @@ def update_attrs_from_base_meta( # noqa: CCR001
|
||||
parent_value=parent_value,
|
||||
)
|
||||
parent_value = [
|
||||
ormar.UniqueColumns(*x._pending_colargs) for x in parent_value
|
||||
ormar.fields.constraints.UniqueColumns(*x._pending_colargs)
|
||||
for x in parent_value
|
||||
]
|
||||
if isinstance(current_value, list):
|
||||
current_value.extend(parent_value)
|
||||
|
||||
Reference in New Issue
Block a user