--- /dev/null
+from __future__ import annotations
+
+from typing import Any
+from typing import Iterable
+from typing import Mapping
+from typing import Optional
+from typing import Union
+
+from ..sql._typing import _DDLColumnArgument
+from ..sql.elements import DQLDMLClauseElement
+from ..sql.schema import ColumnCollectionConstraint
+from ..sql.schema import Index
+
+
+_OnConflictConstraintT = Union[str, ColumnCollectionConstraint, Index, None]
+_OnConflictIndexElementsT = Optional[Iterable[_DDLColumnArgument]]
+_OnConflictIndexWhereT = Optional[DQLDMLClauseElement]
+_OnConflictSetT = Optional[Mapping[Any, Any]]
+_OnConflictWhereT = Union[DQLDMLClauseElement, str, None]
+# mysql/dml.py
# Copyright (C) 2005-2023 the SQLAlchemy authors and contributors
# <see AUTHORS file>
#
# This module is part of SQLAlchemy and is released under
# the MIT License: https://www.opensource.org/licenses/mit-license.php
-# mypy: ignore-errors
+from __future__ import annotations
+from typing import Any
+from typing import List
+from typing import Mapping
+from typing import Optional
+from typing import Tuple
+from typing import Union
from ... import exc
from ... import util
+from ...sql._typing import _DMLTableArgument
from ...sql.base import _exclusive_against
from ...sql.base import _generative
from ...sql.base import ColumnCollection
+from ...sql.base import ReadOnlyColumnCollection
from ...sql.dml import Insert as StandardInsert
from ...sql.elements import ClauseElement
+from ...sql.elements import KeyedColumnElement
from ...sql.expression import alias
+from ...sql.selectable import NamedFromClause
from ...util.typing import Self
__all__ = ("Insert", "insert")
-def insert(table):
+def insert(table: _DMLTableArgument) -> Insert:
"""Construct a MySQL/MariaDB-specific variant :class:`_mysql.Insert`
construct.
inherit_cache = False
@property
- def inserted(self):
+ def inserted(
+ self,
+ ) -> ReadOnlyColumnCollection[str, KeyedColumnElement[Any]]:
"""Provide the "inserted" namespace for an ON DUPLICATE KEY UPDATE
statement
return self.inserted_alias.columns
@util.memoized_property
- def inserted_alias(self):
+ def inserted_alias(self) -> NamedFromClause:
return alias(self.table, name="inserted")
@_generative
"has an ON DUPLICATE KEY clause present"
},
)
- def on_duplicate_key_update(self, *args, **kw) -> Self:
+ def on_duplicate_key_update(self, *args: _UpdateArg, **kw: Any) -> Self:
r"""
Specifies the ON DUPLICATE KEY UPDATE clause.
else:
values = kw
- inserted_alias = getattr(self, "inserted_alias", None)
- self._post_values_clause = OnDuplicateClause(inserted_alias, values)
+ self._post_values_clause = OnDuplicateClause(
+ self.inserted_alias, values
+ )
return self
class OnDuplicateClause(ClauseElement):
__visit_name__ = "on_duplicate_key_update"
- _parameter_ordering = None
+ _parameter_ordering: Optional[List[str]] = None
stringify_dialect = "mysql"
- def __init__(self, inserted_alias, update):
+ def __init__(
+ self, inserted_alias: NamedFromClause, update: _UpdateArg
+ ) -> None:
self.inserted_alias = inserted_alias
# auto-detect that parameters should be ordered. This is copied from
"of a Table object"
)
self.update = update
+
+
+_UpdateArg = Union[
+ Mapping[Any, Any], List[Tuple[str, Any]], ColumnCollection[Any, Any]
+]
+# mysql/mariadb.py
+# Copyright (C) 2005-2023 the SQLAlchemy authors and contributors
+# <see AUTHORS file>
+#
+# This module is part of SQLAlchemy and is released under
+# the MIT License: https://www.opensource.org/licenses/mit-license.php
# mypy: ignore-errors
-
from .base import MariaDBIdentifierPreparer
from .base import MySQLDialect
-# postgresql/on_conflict.py
+# postgresql/dml.py
# Copyright (C) 2005-2023 the SQLAlchemy authors and contributors
# <see AUTHORS file>
#
# This module is part of SQLAlchemy and is released under
# the MIT License: https://www.opensource.org/licenses/mit-license.php
-# mypy: ignore-errors
+from __future__ import annotations
+
+from typing import Any
+from typing import Optional
from . import ext
+from .._typing import _OnConflictConstraintT
+from .._typing import _OnConflictIndexElementsT
+from .._typing import _OnConflictIndexWhereT
+from .._typing import _OnConflictSetT
+from .._typing import _OnConflictWhereT
from ... import util
from ...sql import coercions
from ...sql import roles
from ...sql import schema
+from ...sql._typing import _DMLTableArgument
from ...sql.base import _exclusive_against
from ...sql.base import _generative
from ...sql.base import ColumnCollection
+from ...sql.base import ReadOnlyColumnCollection
from ...sql.dml import Insert as StandardInsert
from ...sql.elements import ClauseElement
+from ...sql.elements import KeyedColumnElement
from ...sql.expression import alias
from ...util.typing import Self
__all__ = ("Insert", "insert")
-def insert(table):
+def insert(table: _DMLTableArgument) -> Insert:
"""Construct a PostgreSQL-specific variant :class:`_postgresql.Insert`
construct.
inherit_cache = False
@util.memoized_property
- def excluded(self):
+ def excluded(
+ self,
+ ) -> ReadOnlyColumnCollection[str, KeyedColumnElement[Any]]:
"""Provide the ``excluded`` namespace for an ON CONFLICT statement
PG's ON CONFLICT clause allows reference to the row that would
@_on_conflict_exclusive
def on_conflict_do_update(
self,
- constraint=None,
- index_elements=None,
- index_where=None,
- set_=None,
- where=None,
+ constraint: _OnConflictConstraintT = None,
+ index_elements: _OnConflictIndexElementsT = None,
+ index_where: _OnConflictIndexWhereT = None,
+ set_: _OnConflictSetT = None,
+ where: _OnConflictWhereT = None,
) -> Self:
r"""
Specifies a DO UPDATE SET action for ON CONFLICT clause.
@_on_conflict_exclusive
def on_conflict_do_nothing(
self,
- constraint=None,
- index_elements=None,
- index_where=None,
+ constraint: _OnConflictConstraintT = None,
+ index_elements: _OnConflictIndexElementsT = None,
+ index_where: _OnConflictIndexWhereT = None,
) -> Self:
"""
Specifies a DO NOTHING action for ON CONFLICT clause.
class OnConflictClause(ClauseElement):
stringify_dialect = "postgresql"
- def __init__(self, constraint=None, index_elements=None, index_where=None):
+ constraint_target: Optional[str]
+ inferred_target_elements: _OnConflictIndexElementsT
+ inferred_target_whereclause: _OnConflictIndexWhereT
+
+ def __init__(
+ self,
+ constraint: _OnConflictConstraintT = None,
+ index_elements: _OnConflictIndexElementsT = None,
+ index_where: _OnConflictIndexWhereT = None,
+ ):
if constraint is not None:
if not isinstance(constraint, str) and isinstance(
constraint,
def __init__(
self,
- constraint=None,
- index_elements=None,
- index_where=None,
- set_=None,
- where=None,
+ constraint: _OnConflictConstraintT = None,
+ index_elements: _OnConflictIndexElementsT = None,
+ index_where: _OnConflictIndexWhereT = None,
+ set_: _OnConflictSetT = None,
+ where: _OnConflictWhereT = None,
):
super().__init__(
constraint=constraint,
+# sqlite/dml.py
# Copyright (C) 2005-2023 the SQLAlchemy authors and contributors
# <see AUTHORS file>
#
# This module is part of SQLAlchemy and is released under
# the MIT License: https://www.opensource.org/licenses/mit-license.php
-# mypy: ignore-errors
+from __future__ import annotations
+from typing import Any
+from .._typing import _OnConflictIndexElementsT
+from .._typing import _OnConflictIndexWhereT
+from .._typing import _OnConflictSetT
+from .._typing import _OnConflictWhereT
from ... import util
from ...sql import coercions
from ...sql import roles
+from ...sql._typing import _DMLTableArgument
from ...sql.base import _exclusive_against
from ...sql.base import _generative
from ...sql.base import ColumnCollection
+from ...sql.base import ReadOnlyColumnCollection
from ...sql.dml import Insert as StandardInsert
from ...sql.elements import ClauseElement
+from ...sql.elements import KeyedColumnElement
from ...sql.expression import alias
from ...util.typing import Self
-
__all__ = ("Insert", "insert")
-def insert(table):
+def insert(table: _DMLTableArgument) -> Insert:
"""Construct a sqlite-specific variant :class:`_sqlite.Insert`
construct.
inherit_cache = False
@util.memoized_property
- def excluded(self):
+ def excluded(
+ self,
+ ) -> ReadOnlyColumnCollection[str, KeyedColumnElement[Any]]:
"""Provide the ``excluded`` namespace for an ON CONFLICT statement
SQLite's ON CONFLICT clause allows reference to the row that would
@_on_conflict_exclusive
def on_conflict_do_update(
self,
- index_elements=None,
- index_where=None,
- set_=None,
- where=None,
+ index_elements: _OnConflictIndexElementsT = None,
+ index_where: _OnConflictIndexWhereT = None,
+ set_: _OnConflictSetT = None,
+ where: _OnConflictWhereT = None,
) -> Self:
r"""
Specifies a DO UPDATE SET action for ON CONFLICT clause.
@_generative
@_on_conflict_exclusive
def on_conflict_do_nothing(
- self, index_elements=None, index_where=None
+ self,
+ index_elements: _OnConflictIndexElementsT = None,
+ index_where: _OnConflictIndexWhereT = None,
) -> Self:
"""
Specifies a DO NOTHING action for ON CONFLICT clause.
class OnConflictClause(ClauseElement):
stringify_dialect = "sqlite"
- def __init__(self, index_elements=None, index_where=None):
+ constraint_target: None
+ inferred_target_elements: _OnConflictIndexElementsT
+ inferred_target_whereclause: _OnConflictIndexWhereT
+
+ def __init__(
+ self,
+ index_elements: _OnConflictIndexElementsT = None,
+ index_where: _OnConflictIndexWhereT = None,
+ ):
if index_elements is not None:
self.constraint_target = None
self.inferred_target_elements = index_elements
def __init__(
self,
- index_elements=None,
- index_where=None,
- set_=None,
- where=None,
+ index_elements: _OnConflictIndexElementsT = None,
+ index_where: _OnConflictIndexWhereT = None,
+ set_: _OnConflictSetT = None,
+ where: _OnConflictWhereT = None,
):
super().__init__(
index_elements=index_elements,
--- /dev/null
+from sqlalchemy import Integer
+from sqlalchemy.dialects.mysql import insert
+from sqlalchemy.orm import DeclarativeBase
+from sqlalchemy.orm import Mapped
+from sqlalchemy.orm import mapped_column
+
+
+class Base(DeclarativeBase):
+ pass
+
+
+class Test(Base):
+ __tablename__ = "test_table_json"
+
+ id = mapped_column(Integer, primary_key=True)
+ data: Mapped[str] = mapped_column()
+
+
+insert(Test).on_duplicate_key_update(
+ {"id": 42, Test.data: 99}, [("foo", 44)], data=99, id="foo"
+).inserted.foo.desc()
from sqlalchemy import or_
from sqlalchemy import select
from sqlalchemy import Text
+from sqlalchemy import UniqueConstraint
from sqlalchemy.dialects.postgresql import ARRAY
from sqlalchemy.dialects.postgresql import array
+from sqlalchemy.dialects.postgresql import insert
from sqlalchemy.dialects.postgresql import JSONB
from sqlalchemy.dialects.postgresql import UUID
from sqlalchemy.orm import DeclarativeBase
# EXPECTED_TYPE: UUID
reveal_type(t1.ident)
+
+unique = UniqueConstraint(name="my_constraint")
+insert(Test).on_conflict_do_nothing(
+ "foo", [Test.id], Test.id > 0
+).on_conflict_do_update(
+ unique, ["foo"], Test.id > 0, {"id": 42, Test.ident: 99}, Test.id == 22
+).excluded.foo.desc()
--- /dev/null
+from sqlalchemy import Integer
+from sqlalchemy import UniqueConstraint
+from sqlalchemy.dialects.sqlite import insert
+from sqlalchemy.orm import DeclarativeBase
+from sqlalchemy.orm import Mapped
+from sqlalchemy.orm import mapped_column
+
+
+class Base(DeclarativeBase):
+ pass
+
+
+class Test(Base):
+ __tablename__ = "test_table_json"
+
+ id = mapped_column(Integer, primary_key=True)
+ data: Mapped[str] = mapped_column()
+
+
+unique = UniqueConstraint(name="my_constraint")
+insert(Test).on_conflict_do_nothing("foo", Test.id > 0).on_conflict_do_update(
+ unique, Test.id > 0, {"id": 42, Test.data: 99}, Test.id == 22
+).excluded.foo.desc()
projects / thirdparty / sqlalchemy / sqlalchemy.git / commitdiff
