# Path: psqlextra/fields/hstore_field.py
# class HStoreField(DjangoHStoreField):
#     """Improved version of Django's :see:HStoreField that adds support for
#     database-level constraints.
# 
#     Notes:
#         - For the implementation of uniqueness, see the
#           custom database back-end.
#     """
# 
#     def __init__(
#         self,
#         *args,
#         uniqueness: Optional[List[Union[str, Tuple[str, ...]]]] = None,
#         required: Optional[List[str]] = None,
#         **kwargs
#     ):
#         """Initializes a new instance of :see:HStoreField.
# 
#         Arguments:
#             uniqueness:
#                 List of keys to enforce as unique. Use tuples
#                 to enforce multiple keys together to be unique.
# 
#             required:
#                 List of keys that should be enforced as required.
#         """
# 
#         super(HStoreField, self).__init__(*args, **kwargs)
# 
#         self.uniqueness = uniqueness
#         self.required = required
# 
#     def get_prep_value(self, value):
#         """Override the base class so it doesn't cast all values to strings.
# 
#         psqlextra supports expressions in hstore fields, so casting all
#         values to strings is a bad idea.
#         """
# 
#         value = Field.get_prep_value(self, value)
# 
#         if isinstance(value, dict):
#             prep_value = {}
#             for key, val in value.items():
#                 if isinstance(val, Expression):
#                     prep_value[key] = val
#                 elif val is not None:
#                     prep_value[key] = str(val)
#                 else:
#                     prep_value[key] = val
# 
#             value = prep_value
# 
#         if isinstance(value, list):
#             value = [str(item) for item in value]
# 
#         return value
# 
#     def deconstruct(self):
#         """Gets the values to pass to :see:__init__ when re-creating this
#         object."""
# 
#         name, path, args, kwargs = super(HStoreField, self).deconstruct()
# 
#         if self.uniqueness is not None:
#             kwargs["uniqueness"] = self.uniqueness
# 
#         if self.required is not None:
#             kwargs["required"] = self.required
# 
#         return name, path, args, kwargs
# 
# Path: psqlextra/query.py
# class PostgresQuerySet(models.QuerySet):
#     def __init__(self, model=None, query=None, using=None, hints=None):
#     def annotate(self, **annotations):
#     def rename_annotations(self, **annotations):
#     def on_conflict(
#         self,
#         fields: ConflictTarget,
#         action: ConflictAction,
#         index_predicate: Optional[Union[Expression, Q, str]] = None,
#         update_condition: Optional[Union[Expression, Q, str]] = None,
#     ):
#     def bulk_insert(
#         self,
#         rows: List[dict],
#         return_model: bool = False,
#         using: Optional[str] = None,
#     ):
#     def insert(self, using: Optional[str] = None, **fields):
#     def insert_and_get(self, using: Optional[str] = None, **fields):
#     def upsert(
#         self,
#         conflict_target: ConflictTarget,
#         fields: dict,
#         index_predicate: Optional[Union[Expression, Q, str]] = None,
#         using: Optional[str] = None,
#         update_condition: Optional[Union[Expression, Q, str]] = None,
#     ) -> int:
#     def upsert_and_get(
#         self,
#         conflict_target: ConflictTarget,
#         fields: dict,
#         index_predicate: Optional[Union[Expression, Q, str]] = None,
#         using: Optional[str] = None,
#         update_condition: Optional[Union[Expression, Q, str]] = None,
#     ):
#     def bulk_upsert(
#         self,
#         conflict_target: ConflictTarget,
#         rows: Iterable[Dict],
#         index_predicate: Optional[Union[Expression, Q, str]] = None,
#         return_model: bool = False,
#         using: Optional[str] = None,
#         update_condition: Optional[Union[Expression, Q, str]] = None,
#     ):
#         def is_empty(r):
#     def _create_model_instance(
#         self, field_values: dict, using: str, apply_converters: bool = True
#     ):
#     def _build_insert_compiler(
#         self, rows: Iterable[Dict], using: Optional[str] = None
#     ):
#     def _is_magical_field(self, model_instance, field, is_insert: bool):
#     def _get_upsert_fields(self, kwargs):
# 
# Path: tests/fake_model.py
# def get_fake_model(fields=None, model_base=PostgresModel, meta_options={}):
#     """Defines a fake model and creates it in the database."""
# 
#     model = define_fake_model(fields, model_base, meta_options)
# 
#     with connection.schema_editor() as schema_editor:
#         schema_editor.create_model(model)
# 
#     return model


import pytest
from django.db import models
from psqlextra.fields import HStoreField
from psqlextra.query import ConflictAction
from .fake_model import get_fake_model

def test_on_conflict_update_foreign_key_by_id():
    """Tests whether simple upsert works correctly when the conflicting field
    is a foreign key specified as an id."""

    other_model = get_fake_model({})

    model = get_fake_model(
        {
            "other": models.OneToOneField(
                other_model, on_delete=models.CASCADE
            ),
            "data": models.CharField(max_length=255),
        }
    )

    other_obj = other_model.objects.create()

    obj1 = model.objects.on_conflict(
        ["other_id"], ConflictAction.UPDATE
    ).insert_and_get(other_id=other_obj.pk, data="some data")

    assert obj1.other == other_obj
    assert obj1.data == "some data"

    obj1.refresh_from_db()
    assert obj1.other == other_obj
    assert obj1.data == "some data"

    obj2 = model.objects.on_conflict(

# Path: psqlextra/fields/hstore_field.py
# class HStoreField(DjangoHStoreField):
#     """Improved version of Django's :see:HStoreField that adds support for
#     database-level constraints.
# 
#     Notes:
#         - For the implementation of uniqueness, see the
#           custom database back-end.
#     """
# 
#     def __init__(
#         self,
#         *args,
#         uniqueness: Optional[List[Union[str, Tuple[str, ...]]]] = None,
#         required: Optional[List[str]] = None,
#         **kwargs
#     ):
#         """Initializes a new instance of :see:HStoreField.
# 
#         Arguments:
#             uniqueness:
#                 List of keys to enforce as unique. Use tuples
#                 to enforce multiple keys together to be unique.
# 
#             required:
#                 List of keys that should be enforced as required.
#         """
# 
#         super(HStoreField, self).__init__(*args, **kwargs)
# 
#         self.uniqueness = uniqueness
#         self.required = required
# 
#     def get_prep_value(self, value):
#         """Override the base class so it doesn't cast all values to strings.
# 
#         psqlextra supports expressions in hstore fields, so casting all
#         values to strings is a bad idea.
#         """
# 
#         value = Field.get_prep_value(self, value)
# 
#         if isinstance(value, dict):
#             prep_value = {}
#             for key, val in value.items():
#                 if isinstance(val, Expression):
#                     prep_value[key] = val
#                 elif val is not None:
#                     prep_value[key] = str(val)
#                 else:
#                     prep_value[key] = val
# 
#             value = prep_value
# 
#         if isinstance(value, list):
#             value = [str(item) for item in value]
# 
#         return value
# 
#     def deconstruct(self):
#         """Gets the values to pass to :see:__init__ when re-creating this
#         object."""
# 
#         name, path, args, kwargs = super(HStoreField, self).deconstruct()
# 
#         if self.uniqueness is not None:
#             kwargs["uniqueness"] = self.uniqueness
# 
#         if self.required is not None:
#             kwargs["required"] = self.required
# 
#         return name, path, args, kwargs
# 
# Path: psqlextra/query.py
# class PostgresQuerySet(models.QuerySet):
#     def __init__(self, model=None, query=None, using=None, hints=None):
#     def annotate(self, **annotations):
#     def rename_annotations(self, **annotations):
#     def on_conflict(
#         self,
#         fields: ConflictTarget,
#         action: ConflictAction,
#         index_predicate: Optional[Union[Expression, Q, str]] = None,
#         update_condition: Optional[Union[Expression, Q, str]] = None,
#     ):
#     def bulk_insert(
#         self,
#         rows: List[dict],
#         return_model: bool = False,
#         using: Optional[str] = None,
#     ):
#     def insert(self, using: Optional[str] = None, **fields):
#     def insert_and_get(self, using: Optional[str] = None, **fields):
#     def upsert(
#         self,
#         conflict_target: ConflictTarget,
#         fields: dict,
#         index_predicate: Optional[Union[Expression, Q, str]] = None,
#         using: Optional[str] = None,
#         update_condition: Optional[Union[Expression, Q, str]] = None,
#     ) -> int:
#     def upsert_and_get(
#         self,
#         conflict_target: ConflictTarget,
#         fields: dict,
#         index_predicate: Optional[Union[Expression, Q, str]] = None,
#         using: Optional[str] = None,
#         update_condition: Optional[Union[Expression, Q, str]] = None,
#     ):
#     def bulk_upsert(
#         self,
#         conflict_target: ConflictTarget,
#         rows: Iterable[Dict],
#         index_predicate: Optional[Union[Expression, Q, str]] = None,
#         return_model: bool = False,
#         using: Optional[str] = None,
#         update_condition: Optional[Union[Expression, Q, str]] = None,
#     ):
#         def is_empty(r):
#     def _create_model_instance(
#         self, field_values: dict, using: str, apply_converters: bool = True
#     ):
#     def _build_insert_compiler(
#         self, rows: Iterable[Dict], using: Optional[str] = None
#     ):
#     def _is_magical_field(self, model_instance, field, is_insert: bool):
#     def _get_upsert_fields(self, kwargs):
# 
# Path: tests/fake_model.py
# def get_fake_model(fields=None, model_base=PostgresModel, meta_options={}):
#     """Defines a fake model and creates it in the database."""
# 
#     model = define_fake_model(fields, model_base, meta_options)
# 
#     with connection.schema_editor() as schema_editor:
#         schema_editor.create_model(model)
# 
#     return model


# Path: tests/test_on_conflict_update.py
import pytest
from django.db import models
from psqlextra.fields import HStoreField
from psqlextra.query import ConflictAction
from .fake_model import get_fake_model


def test_on_conflict_update_foreign_key_by_id():
    """Tests whether simple upsert works correctly when the conflicting field
    is a foreign key specified as an id."""

    other_model = get_fake_model({})

    model = get_fake_model(
        {
            "other": models.OneToOneField(
                other_model, on_delete=models.CASCADE
            ),
            "data": models.CharField(max_length=255),
        }
    )

    other_obj = other_model.objects.create()

    obj1 = model.objects.on_conflict(
        ["other_id"], ConflictAction.UPDATE
    ).insert_and_get(other_id=other_obj.pk, data="some data")

    assert obj1.other == other_obj
    assert obj1.data == "some data"

    obj1.refresh_from_db()
    assert obj1.other == other_obj
    assert obj1.data == "some data"

    obj2 = model.objects.on_conflict(

        ["other_id"], ConflictAction.UPDATE