我正在尝试使用 attrs 结构和解构在 django 中使用石墨烯创建一个人模式，以便使用 JSONB 将我的数据输入到数据库中。
到目前为止，这是我创建的：
模型.py：
from django.db import models
# Create your models here.
class PersonModel(models.Model):
data = models.JSONField(null=True)
架构.py：
# from typing_extensions import Required
import graphene
import cattr
from graphene_django.types import DjangoObjectType
from graphene import Mutation, InputObjectType,ObjectType, String, Boolean, Field, Int, List, NonNull, ID, Argument
from .models import PersonModel
from .person import PersonDataClass
class PersonSchema(DjangoObjectType):
id = Int,
name = String,
age = Int,
address_one = String,
address_two = String
class Meta:
model = PersonModel
class PersonSchemaOutput(PersonSchema, DjangoObjectType):
# notice we only need ID in output and not in input of Mutation of Create
id = graphene.ID
class Meta:
model = PersonModel
pass
class PersonSchemaInputCreate(PersonSchema, InputObjectType):
id = graphene.ID
class Meta:
model = PersonModel
pass
class PersonSchemaInputUpdate(InputObjectType):
# lets say we only allow update of following
class Arguments:
id = graphene.ID
name = String,
age = Int,
class Meta:
model = PersonModel
pass
class CreatePerson(graphene.Mutation):
ok = Boolean()
class Arguments:
first_name = graphene.String()
age = graphene.Int()
address_one = graphene.String()
address_two = graphene.String()
person_data = Argument(PersonSchemaInputCreate())
# output will be the created person
person = Argument(PersonSchemaOutput)
ok = graphene.Boolean()
@classmethod
def mutate(self, info, person_data):
person_data_class = PersonDataClass(name=person_data.name, age=person_data.age, address_one=person_data.address_one, address_two=person_data.address_two)
# now create json_data
person_json_data = cattr.unstructure(person_data_class)
# create entry in DB
person_db_record = PersonModel(data=person_json_data)
person_db_record.save()
person_instance = PersonSchemaOutput(
id = person_db_record.id,
name= person_db_record.data["name"],
age=person_db_record.data["age"],
address_one=person_db_record.data["address_one"],
address_two=person_db_record.data["address_two"],
)
return CreatePerson(person=person_instance)
def resolve_all_persons(self, info, **kwargs):
persons_set = PersonModel.objects.all()
# python shortcut for arrays
return [
PersonSchemaOutput(
id = person_db_record.id,
name= person_db_record.data["name"],
age=person_db_record.data["age"],
address_one=person_db_record.data["address_one"],
address_two=person_db_record.data["address_two"])
for person_db_record in persons_set
]
def resolve_person(self, info, person_id):
person_db_record = PersonModel.objects.get(pk=person_id)
return PersonSchemaOutput(
id = person_db_record.id,
name= person_db_record.data["name"],
age=person_db_record.data["age"],
address_one=person_db_record.data["address_one"],
address_two=person_db_record.data["address_two"],
)
class Query(object):
all_Persons = graphene.List(PersonSchema)
person = Field(PersonSchema)
ok = Boolean()
@classmethod
def mutate(cls, root, info, text, id):
person = PersonSchema.objects.get(pk=id)
person.text = text
person.save()
# Notice we return an instance of this mutation
return Query(person=person)
def resolve_all_Persons(self, info, **kwargs):
person_set = PersonModel.objects.all()
result = []
for person_db in person_set:
person_data_class = cattr.structure(person_db, PersonModel)
result.append(PersonSchema)
class UpdatePerson(graphene.Mutation):
ok = graphene.Boolean()
class Arguments:
person_data = PersonSchemaInputUpdate()
person = Argument(PersonSchemaOutput)
@classmethod
def mutate(root, info, person_data=None):
ok = True
person_db_record = PersonModel.objects.get(pk=person_data.id)
if person_db_record:
person_data_class = cattr.structure(person_db_record.data, PersonDataClass)
# this is how you update values within a dataclass attr object and if any validation fails this will error out
# right now we are
person_data_class = cattr.evolve(person_data_class, name=person_data.name,age=person_data.age)
# no error on prev step means attr validation passed
# now update the DB object for the specific record
person_new_data_json = cattr.unstructure(person_data_class)
person_db_record.data= person_new_data_json
person_db_record.save()
person_instance = PersonSchemaOutput(
id = person_db_record.id,
name= person_db_record.data["name"],
age= person_db_record.data["age"],
address_one=person_db_record.data["address_one"],
address_two=person_db_record.data["address_two"],
)
return UpdatePerson(person=person_instance, ok=ok)
return UpdatePerson(person=None, ok=ok)
class Mutation(graphene.ObjectType):
create_person = CreatePerson.Field()
update_person = UpdatePerson.Field()
和我的 person.py （目前没有给我任何问题）：
from attr import attrs, attrib
import cattr
from attr.validators import (
instance_of,
optional,
)
from typing import Optional
@attrs
class PersonDataClass():
name: Optional[str] = attrib(default=None, validator=optional(instance_of(str)))
age: Optional[str] = attrib(default=None, validator=optional(instance_of(int)))
address_one: Optional[str] = attrib(default=None, validator=optional(instance_of(str)))
address_two: Optional[str] = attrib(default=None, validator=optional(instance_of(str)))
@age.validator
def _check_age(self, _, value):
if value <=20:
raise ValueError("age must be greater than 20")
这是我现在遇到的错误
AssertionError: PersonSchemaInputUpdate fields must be a mapping (dict / OrderedDict) with field names as keys or a function which returns such a mapping.
在整个构建过程中，我遇到了很多此类错误，并且根据我对 Django 和 Graphene 的介绍知识，我确信我主要只是在代码上添加了创可贴，因此您可以建议进行任何重构来实现这一点更坚固将不胜感激。