Skip to content

Instantly share code, notes, and snippets.

@MrHen MrHen/serializers.py Secret

Last active Sep 17, 2019
Embed
What would you like to do?
DRF Remote Serializer
from rest_framework import serializers
from auth_service.client import AuthClient
from .serializers_remote import RemoteSerializer
auth_client = AuthClient()
class RemoteUserSerializer(RemoteSerializer):
id = serializers.IntegerField(min_value=0, read_only=True)
organization_id = serializers.IntegerField(min_value=0)
is_active = serializers.BooleanField(default=True, required=False)
email = serializers.EmailField()
first_name = serializers.CharField(
max_length=100,
required=False,
allow_null=True,
allow_blank=True,
)
last_name = serializers.CharField(
max_length=100,
required=False,
allow_null=True,
allow_blank=True,
)
full_name = serializers.CharField(
max_length=201,
required=False,
allow_null=True,
allow_blank=True,
)
original_id = serializers.IntegerField(min_value=0)
password = serializers.CharField(write_only=True)
roles = serializers.ListField(
child=serializers.IntegerField(
min_value=0,
),
required=False,
)
class Meta:
base_url = 'users_raw'
client = auth_client
import json
import logging
from django.core.serializers.json import DjangoJSONEncoder
from requests.exceptions import HTTPError
from rest_framework import serializers
from auth_service.client import AuthClient
logger = logging.getLogger(__name__)
auth_client = AuthClient()
class RemoteSerializer(serializers.Serializer):
def get_base_url(self):
meta = getattr(self, 'Meta', None)
base_url = getattr(meta, 'base_url', None)
return base_url
def get_client(self):
meta = getattr(self, 'Meta', None)
client = getattr(meta, 'client', None)
if not client:
client = auth_client
return client
def remote_load(self, instance):
id = instance.get('id')
if not id:
raise serializers.ValidationError('RemoteSerializer requires id')
client = self.get_client()
base_url = self.get_base_url()
response = client.get(f'{base_url}/{id}/')
response_status = response.status_code
response_json = response.json()
try:
response.raise_for_status()
except HTTPError:
logger.error(
'RemoteSerializer.remote_load GET failure'
f', response_status={response_status}'
f', response_json={response_json}'
)
if response_status == 404:
raise serializers.ValidationError(f'Could not find remote user with id={id}')
else:
# TODO: Handle errors better
raise serializers.ValidationError(f'Unknown error retrieving remote user')
return response_json
def remote_create(self, validated_data):
client = self.get_client()
base_url = self.get_base_url()
data = json.dumps(validated_data, cls=DjangoJSONEncoder)
response = client.post(f'{base_url}/', data=data)
response_status = response.status_code
response_json = response.json()
try:
response.raise_for_status()
except HTTPError:
logger.error(
'RemoteUserSerializer.remote_create POST failure'
f', response_status={response_status}'
f', response_json={response_json}'
)
if response_status == 404:
raise serializers.ValidationError(f'Could not find remote user with id={id}')
else:
# TODO: Handle errors better
raise serializers.ValidationError(f'Unknown error creating remote user')
return response_json
def remote_update(self, instance, validated_data):
client = self.get_client()
base_url = self.get_base_url()
remote_user_id = instance.get('id')
if not remote_user_id:
raise serializers.ValidationError(f'Remote user id is required')
data = json.dumps(validated_data, cls=DjangoJSONEncoder)
response = client.patch(f'{base_url}/{remote_user_id}/', data=data)
response_status = response.status_code
response_json = response.json()
try:
response.raise_for_status()
except HTTPError:
logger.error(
'RemoteUserSerializer.remote_update POST failure'
f', response_status={response_status}'
f', response_json={response_json}'
)
if response_status == 404:
raise serializers.ValidationError(f'Could not find remote user with id={id}')
else:
# TODO: Handle errors better
raise serializers.ValidationError(f'Unknown error creating remote user')
return response_json
def to_representation(self, instance):
remote_json = self.remote_load(instance)
return super().to_representation(remote_json)
def create(self, validated_data):
return self.remote_create(validated_data)
def update(self, instance, validated_data):
return self.remote_update(instance, validated_data)
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment
You can’t perform that action at this time.