-
-
Save MrHen/6e96f2aa2729446c15ed4ef4189a087b to your computer and use it in GitHub Desktop.
DRF Remote Serializer
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
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 |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
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