Published in
7 min readApr 18, 2018
Refactor, Componentization!
Teknik yang saya gunakan adalah Componentization, yang secara definisi yang saya ambil dari https://en.wikipedia.org/wiki/Code_refactoring sebagai berikut, Componentization breaks code down into reusable semantic units that present clear, well-defined, simple-to-use interfaces. Saya ingin menghilangkan duplicate code yang ada pada sistem suling, dan juga membuatnya mudah digunakan oleh banyak orang, dan mudah untuk proses peremajaan.
import json
import requests
import os
import logging
import httplib2
from django.contrib.auth.models import User
from django.core.exceptions import (
PermissionDenied, SuspiciousOperation
)
from django.http import (
HttpRequest, HttpResponseRedirect,
HttpResponse, JsonResponse,
HttpResponseBadRequest
)
from django.shortcuts import render
from django.contrib.auth.decorators import login_required
from django.shortcuts import render
from django.conf import settings
from rest_framework.views import APIView
from rest_framework.response import Response
from rest_framework import permissions
from rest_framework.generics import (
ListAPIView,
)
from rest_framework import status
from django.conf import settings
from django.shortcuts import render
from django.http import (
HttpRequest, HttpResponseRedirect, HttpResponse,
JsonResponse
)
from django.contrib.auth import (
login as auth_login,
logout as auth_logout
)
from app_auth.utils import AuthHelper, get_env, get_random_state
from app_auth.models import UserMahasiswa, UserPerusahaan, User
from app_auth.serializers import (
UserMahasiswaSerializer,
UserPerusahaanSerializer
)
from app_auth.utils import (
AuthHelper,
LinkedinHelper,
JWTHelper
)
from googleapiclient.discovery import build
from oauth2client.client import flow_from_clientsecrets
class UserMahasiswaAPIView(ListAPIView):
queryset = UserMahasiswa.objects.all()
serializer_class = UserMahasiswaSerializer
permission_classes = (permissions.IsAuthenticated, )
class LoginMahasiswaAPIView(APIView):
permission_classes = (permissions.AllowAny, )
def post(self, request, format=None):
try:
__USERNAME__ = request.data['username']
__PASSWORD__ = request.data['password']
helper = AuthHelper()
access_token = helper.get_access_token(__USERNAME__, __PASSWORD__)
verified_user = helper.verify_user(access_token)
if verified_user.status_code == status.HTTP_200_OK:
verified_user = verified_user.json()
if verified_user['role'] not in settings.VERIFIED_ROLE:
message = {
'message': 'Permission Denied'
}
return Response(message,
status=status.HTTP_400_BAD_REQUEST)
else:
message = {
'message': 'Bad Request to API CS'
}
return Response(message, status=status.HTTP_400_BAD_REQUEST)
user_mahasiswa_id = verified_user['identity_number']
username = verified_user['username']
role = verified_user['role']
email = helper.get_email_by_username(username)
user_mahasiswa = None
try:
user = User.objects.get(
email=email)
user_mahasiswa = UserMahasiswa.objects.get(user_id=user.id)
except User.DoesNotExist:
user_csauth = helper.get_user_data(
access_token, user_mahasiswa_id)
fullname = user_csauth['nama']
firstname, lastname = helper.get_firstname_lastname(fullname)
npm = user_csauth['npm']
angkatan = helper.get_angkatan_by_npm(npm)
user = User()
user.username = username
user.email = email
user.first_name = firstname
user.last_name = lastname
user.save()
user_mahasiswa = UserMahasiswa()
user_mahasiswa.user = user
user_mahasiswa.npm = npm
user_mahasiswa.nama_lengkap = fullname
user_mahasiswa.angkatan = angkatan
user_mahasiswa.is_valid = True
user_mahasiswa.save()
# serializer = UserMahasiswaSerializer(user_mahasiswa)
# make user is authenticated
auth_login(request, user)
redirect_uri = os.getenv('FE_URL_V1', '') + "mahasiswa"
response = HttpResponseRedirect(
redirect_uri)
response.set_cookie("user_id", user.id)
response.set_cookie("username", user.username)
response = JWTHelper().store_to_cookies(response, user, 3)
return response
except Exception:
message = {
'message': 'Bad Request'
}
return Response(message, status=status.HTTP_400_BAD_REQUEST)
class LoginPerusahaanAPIView(APIView):
permission_classes = (permissions.AllowAny, )
def authenticate(self, username, password):
try:
user = User.objects.get(username=username)
if user.password == password:
return user
else:
message = {
'message': 'Wrong Password'
}
return Response(message, status=status.HTTP_400_BAD_REQUEST)
except User.DoesNotExist:
message = {
'message': 'Username Not Found'
}
return Response(message, status=status.HTTP_400_BAD_REQUEST)
def post(self, request, format=None):
__USERNAME__ = request.data['username']
__PASSWORD__ = request.data['password']
user = self.authenticate(__USERNAME__, __PASSWORD__)
user_perusahaan = UserPerusahaan.objects.get(user=user.id)
serializer = UserPerusahaanSerializer(user_perusahaan)
auth_login(request, user_perusahaan)
return Response(serializer.data, status=status.HTTP_200_OK)
class LogoutAPIView(APIView):
def get(self, request, *args, **kwargs):
try:
response = Response()
response.delete_cookie('jwt_')
response.delete_cookie('user_id')
response.delete_cookie('username')
except Exception:
pass
auth_logout(request)
message = {
'message': 'Successfully logged out'
}
response.message = message
response.status = status.HTTP_200_OK
return response
class LinkedinRequestAPIView(APIView):
permission_classes = (permissions.AllowAny,)
def get(self, request, format=None):
CLIENT_ID_LINKEDIN = os.getenv('CLIENT_ID_LINKEDIN', '')
request.session['state'] = LinkedinHelper().get_random_state()
BE_URL = os.getenv('BE_URL', '')
REDIRECT_URI = BE_URL + "auth/linkedin/callback"
uri = (
"https://www.linkedin.com/oauth/v2/authorization?" +
"response_type=code&" +
"client_id=" + CLIENT_ID_LINKEDIN + "&" +
"redirect_uri=" + REDIRECT_URI + "&" +
"state=" + request.session['state'] + "&" +
"scope=r_basicprofile r_emailaddress rw_company_admin w_share"
)
return HttpResponseRedirect(uri)
class LinkedinCallbackAPIView(APIView):
permission_classes = (permissions.AllowAny,)
def get(self, request, format=None):
BE_URL = os.getenv('BE_URL', '')
CLIENT_ID_LINKEDIN = os.getenv('CLIENT_ID_LINKEDIN', '')
CLIENT_SECRET_LINKEDIN = os.getenv(
'CLIENT_SECRET_LINKEDIN', ''
)
REDIRECT_URI = BE_URL + "auth/linkedin/callback"
request_token_uri = "https://www.linkedin.com/oauth/v2/accessToken"
error = request.GET.get('error', '')
if error != '':
return HttpResponse(status=401)
state = request.GET.get('state', '')
if state != request.session['state']:
return HttpResponse(status=403)
code = request.GET.get('code', '')
request_data = {
"grant_type": "authorization_code",
"code": code,
"redirect_uri": REDIRECT_URI,
"client_id": CLIENT_ID_LINKEDIN,
"client_secret": CLIENT_SECRET_LINKEDIN
}
request_header = {
'content_type': "application/json"
}
response = requests.request(
"POST", request_token_uri,
data=request_data,
headers=request_header)
response_data = json.loads(response.content.decode('utf8'))
token = response_data['access_token']
url = (
"https://api.linkedin.com/v1/people/~:(" +
"id,firstName,lastName,email-address)" +
"?oauth2_access_token=" + token + "&format=json"
)
user_data = requests.request("GET", url=url)
user_data = json.loads(user_data.content.decode('utf8'))
__PASSWORD__ = User.objects.make_random_password()
__EMAIL__ = user_data["emailAddress"]
__FIRST_NAME__ = user_data["firstName"]
__LAST_NAME__ = user_data["lastName"]
__FULL_NAME__ = "{} {}".format(__FIRST_NAME__, __LAST_NAME__)
try:
user = User.objects.get(email=__EMAIL__)
except User.DoesNotExist:
user = User()
user.username = __EMAIL__
user.email = __EMAIL__
user.first_name = __FIRST_NAME__
user.last_name = __LAST_NAME__
user.password = __PASSWORD__
user.save()
user_mahasiswa = UserMahasiswa()
user_mahasiswa.user = user
user_mahasiswa.nama_lengkap = __FULL_NAME__
user_mahasiswa.is_valid = False
user_mahasiswa.save()
# make user is authenticated
auth_login(request, user)
redirect_uri = os.getenv('FE_URL_V1', '') + "mahasiswa"
response = HttpResponseRedirect(redirect_uri)
response.set_cookie("user_id", user.id)
response.set_cookie("username", user.username)
response = JWTHelper().store_to_cookies(response, user, 3)
return response
class GoogleAuthBase():
def get_flow(self):
redirect_uri = os.getenv('BE_URL', '') + 'auth/google/callback'
FLOW = flow_from_clientsecrets(
settings.GOOGLE_OAUTH2_CLIENT_SECRETS_JSON,
scope=[
'https://www.googleapis.com/auth/plus.login',
'https://www.googleapis.com/auth/plus.profile.emails.read',
'https://www.googleapis.com/auth/userinfo.profile'
],
redirect_uri=redirect_uri)
return FLOW
class GoogleRequestAuthAPIView(APIView):
permission_classes = (permissions.AllowAny,)
def get(self, request, format=None):
FLOW = GoogleAuthBase().get_flow()
authorize_url = FLOW.step1_get_authorize_url()
return HttpResponseRedirect(authorize_url)
class GoogleCallbackAuthAPIView(APIView):
permission_classes = (permissions.AllowAny,)
def get(self, request, format=None):
if "error" in request.get_full_path():
return HttpResponseBadRequest()
FLOW = GoogleAuthBase().get_flow()
credential = FLOW.step2_exchange(request.GET)
http = httplib2.Http()
http = credential.authorize(http)
service = build("plus", "v1", http=http)
people_resource = service.people()
people_document = people_resource.get(userId='me').execute()
__PASSWORD__ = User.objects.make_random_password()
__EMAIL__ = people_document['emails'][0]['value']
__FIRST_NAME__ = people_document['name']['givenName']
__LAST_NAME__ = people_document['name']['familyName']
__FULL_NAME__ = "{} {}".format(__FIRST_NAME__, __LAST_NAME__)
try:
user = User.objects.get(email=__EMAIL__)
except User.DoesNotExist:
user = User()
user.username = __EMAIL__
user.email = __EMAIL__
user.first_name = __FIRST_NAME__
user.last_name = __LAST_NAME__
user.password = __PASSWORD__
user.save()
user_mahasiswa = UserMahasiswa()
user_mahasiswa.user = user
user_mahasiswa.nama_lengkap = __FULL_NAME__
user_mahasiswa.is_valid = False
user_mahasiswa.save()
# make user is authenticated
auth_login(request, user)
redirect_uri = os.getenv('FE_URL_V1', '') + "mahasiswa"
response = HttpResponseRedirect(redirect_uri)
response.set_cookie("user_id", user.id)
response.set_cookie("username", user.username)
response = JWTHelper().store_to_cookies(response, user, 3)
return responseJika kita liat, ada banyak logic yang digunakan berulang untuk menyimpan user_mahasiswa kedalam database. Oleh karena dibuat satu fungsi untuk dapat dipanggil didalam class-class diatas. Dan fungsi tersebut kami buat di utils.
def save_user_profile(self, email, first_name, last_name):
if email is None:
raise ValueError("Email tidak boleh kosong!")
password = User.objects.make_random_password()
user = User()
user.username = email
user.email = email
user.first_name = first_name
user.last_name = last_name
user.password = password
user.save()
return user
def save_user_mahasiswa_profile(self,
user,
npm,
fullname,
angkatan,
is_valid=False):
if user is None:
raise ValueError("Tidak ada user yang sedang aktif!")
if fullname is None:
raise ValueError("Nama lengkap tidak boleh kosong!")
user_mahasiswa = UserMahasiswa()
user_mahasiswa.user = user
user_mahasiswa.npm = npm
user_mahasiswa.nama_lengkap = fullname
user_mahasiswa.angkatan = angkatan
user_mahasiswa.is_valid = is_valid
user_mahasiswa.save()
return user_mahasiswa
def save_user_linkedin_profile(self, user, first_name, last_name,
headline, location, industry,
current_share, num_connections,
summary, specialties, picture_url,
public_profile_url, email_address):
if user is None:
raise ValueError("Tidak ada user yang sedang aktif!")
user_linkedin_profile = UserLinkedinProfile()
user_linkedin_profile.user = user
user_linkedin_profile.first_name = first_name
user_linkedin_profile.last_name = last_name
user_linkedin_profile.headline = headline
user_linkedin_profile.location = location
user_linkedin_profile.industry = industry
user_linkedin_profile.current_share = current_share
user_linkedin_profile.num_connections = num_connections
user_linkedin_profile.summary = summary
user_linkedin_profile.specialties = specialties
user_linkedin_profile.picture_url = picture_url
user_linkedin_profile.public_profile_url = public_profile_url
user_linkedin_profile.email_address = email_address
user_linkedin_profile.save()
return user_linkedin_profileSehingga views terbaru kami menjadi seperti berikut.
import json
import os
import httplib2
import requests
# Import from django
from django.contrib.auth.models import User
from django.contrib.auth import (
login as auth_login,
logout as auth_logout
)
from django.http import (
HttpResponseRedirect,
HttpResponse, HttpResponseBadRequest
)
from django.conf import settings
# Import from django rest
from rest_framework import (
permissions, status
)
from rest_framework.generics import (
ListAPIView,
)
from rest_framework.response import Response
from rest_framework.views import APIView
# Import from projects
from app_auth.models import UserMahasiswa, UserPerusahaan
from app_auth.serializers import (
UserMahasiswaSerializer,
UserPerusahaanSerializer
)
from app_auth.utils import (
AuthHelper,
LinkedinHelper,
JWTHelper
)
# Import others
from googleapiclient.discovery import build
from oauth2client.client import flow_from_clientsecrets
class UserMahasiswaAPIView(ListAPIView):
queryset = UserMahasiswa.objects.all()
serializer_class = UserMahasiswaSerializer
permission_classes = (permissions.IsAuthenticated,)
class LogoutAPIView(APIView):
def get(self, request, *args, **kwargs):
response = Response()
response.delete_cookie('jwt_')
response.delete_cookie('user_id')
response.delete_cookie('username')
auth_logout(request)
response.message = {
'message': 'You has been successfully logged out.'
}
response.status = status.HTTP_200_OK
return response
class LoginMahasiswaAPIView(APIView):
permission_classes = (permissions.AllowAny,)
def post(self, request, format=None):
try:
__USERNAME__ = request.data['username']
__PASSWORD__ = request.data['password']
helper = AuthHelper()
access_token = helper.get_access_token(__USERNAME__, __PASSWORD__)
verified_user = helper.verify_user(access_token)
if verified_user.status_code == status.HTTP_200_OK:
verified_user = verified_user.json()
if verified_user['role'] not in settings.VERIFIED_ROLE:
message = {
'message': 'Permission Denied'
}
return Response(message,
status=status.HTTP_400_BAD_REQUEST)
else:
message = {
'message': 'Bad Request to API CS'
}
return Response(message, status=status.HTTP_400_BAD_REQUEST)
user_mahasiswa_id = verified_user['identity_number']
username = verified_user['username']
role = verified_user['role']
email = helper.get_email_by_username(username)
user_mahasiswa = None
try:
user = User.objects.get(
email=email)
user_mahasiswa = UserMahasiswa.objects.get(user_id=user.id)
except User.DoesNotExist:
user_csauth = helper.get_user_data(
access_token, user_mahasiswa_id)
fullname = user_csauth['nama']
firstname, lastname = helper.get_firstname_lastname(fullname)
npm = user_csauth['npm']
angkatan = helper.get_angkatan_by_npm(npm)
user = helper.save_user_profile(email, firstname, lastname)
user_mahasiswa = helper.save_user_mahasiswa_profile(
user, npm, fullname, angkatan, True)
# serializer = UserMahasiswaSerializer(user_mahasiswa)
# make user is authenticated
auth_login(request, user)
redirect_uri = os.getenv('FE_URL_V1', '') + "mahasiswa"
response = HttpResponseRedirect(
redirect_uri)
response.set_cookie("user_id", user.id)
response.set_cookie("username", user.username)
response = JWTHelper().store_to_cookies(response, user, 3)
return response
except Exception:
message = {
'message': 'Bad Request'
}
return Response(message, status=status.HTTP_400_BAD_REQUEST)
class LoginPerusahaanAPIView(APIView):
permission_classes = (permissions.AllowAny,)
def authenticate(self, username, password):
try:
user = User.objects.get(username=username)
if user.password == password:
return user
else:
message = {
'message': 'Wrong Password'
}
return Response(message, status=status.HTTP_400_BAD_REQUEST)
except User.DoesNotExist:
message = {
'message': 'Username Not Found'
}
return Response(message, status=status.HTTP_400_BAD_REQUEST)
def post(self, request, format=None):
__USERNAME__ = request.data['username']
__PASSWORD__ = request.data['password']
user = self.authenticate(__USERNAME__, __PASSWORD__)
user_perusahaan = UserPerusahaan.objects.get(user=user.id)
serializer = UserPerusahaanSerializer(user_perusahaan)
auth_login(request, user_perusahaan)
return Response(serializer.data, status=status.HTTP_200_OK)
class LinkedinRequestAPIView(APIView):
permission_classes = (permissions.AllowAny,)
def get(self, request, format=None):
CLIENT_ID_LINKEDIN = os.getenv('CLIENT_ID_LINKEDIN', '')
request.session['state'] = LinkedinHelper().get_random_state()
BE_URL = os.getenv('BE_URL', '')
REDIRECT_URI = BE_URL + "auth/linkedin/callback"
uri = (
"https://www.linkedin.com/oauth/v2/authorization?" +
"response_type=code&" +
"client_id=" + CLIENT_ID_LINKEDIN + "&" +
"redirect_uri=" + REDIRECT_URI + "&" +
"state=" + request.session['state'] + "&" +
"scope=r_basicprofile r_emailaddress rw_company_admin w_share"
)
return HttpResponseRedirect(uri)
class LinkedinCallbackAPIView(APIView):
permission_classes = (permissions.AllowAny,)
def get(self, request, format=None):
BE_URL = os.getenv('BE_URL', '')
CLIENT_ID_LINKEDIN = os.getenv('CLIENT_ID_LINKEDIN', '')
CLIENT_SECRET_LINKEDIN = os.getenv(
'CLIENT_SECRET_LINKEDIN', ''
)
REDIRECT_URI = BE_URL + "auth/linkedin/callback"
request_token_uri = "https://www.linkedin.com/oauth/v2/accessToken"
error = request.GET.get('error', '')
if error != '':
return HttpResponse(status=401)
state = request.GET.get('state', '')
if state != request.session['state']:
return HttpResponse(status=403)
code = request.GET.get('code', '')
request_data = {
"grant_type": "authorization_code",
"code": code,
"redirect_uri": REDIRECT_URI,
"client_id": CLIENT_ID_LINKEDIN,
"client_secret": CLIENT_SECRET_LINKEDIN
}
request_header = {
'content_type': "application/json"
}
response = requests.request(
"POST", request_token_uri,
data=request_data,
headers=request_header)
response_data = json.loads(response.content.decode('utf8'))
token = response_data['access_token']
scope = (
"first-name,last-name,headline,location,industry,"
"current-share,num-connections,summary,specialties,"
"positions,picture-url,public-profile-url,email-address"
)
url = (
"https://api.linkedin.com/v1/people/~:({})?".format(scope)
+ "oauth2_access_token={}&format=json".format(token)
)
user_data = requests.request("GET", url=url)
user_data = json.loads(user_data.content.decode('utf8'))
email = user_data["emailAddress"]
firstname = user_data["firstName"]
lastname = user_data["lastName"]
fullname = "{} {}".format(firstname, lastname)
try:
user = User.objects.get(email=email)
except User.DoesNotExist:
user = AuthHelper().save_user_profile(email, firstname, lastname)
AuthHelper().save_user_mahasiswa_profile(user,
None, fullname, None)
# make user is authenticated
auth_login(request, user)
redirect_uri = os.getenv('FE_URL_V1', '') + "mahasiswa"
response = HttpResponseRedirect(redirect_uri)
response.set_cookie("user_id", user.id)
response.set_cookie("username", user.username)
response = JWTHelper().store_to_cookies(response, user, 3)
return response
class GoogleAuthBase():
def get_flow(self):
redirect_uri = os.getenv('BE_URL', '') + 'auth/google/callback'
FLOW = flow_from_clientsecrets(
settings.GOOGLE_OAUTH2_CLIENT_SECRETS_JSON,
scope=[
'https://www.googleapis.com/auth/plus.login',
'https://www.googleapis.com/auth/plus.profile.emails.read',
'https://www.googleapis.com/auth/userinfo.profile'
],
redirect_uri=redirect_uri)
return FLOW
class GoogleRequestAuthAPIView(APIView):
permission_classes = (permissions.AllowAny,)
def get(self, request, format=None):
FLOW = GoogleAuthBase().get_flow()
authorize_url = FLOW.step1_get_authorize_url()
return HttpResponseRedirect(authorize_url)
class GoogleCallbackAuthAPIView(APIView):
permission_classes = (permissions.AllowAny,)
def get(self, request, format=None):
if "error" in request.get_full_path():
return HttpResponseBadRequest()
FLOW = GoogleAuthBase().get_flow()
credential = FLOW.step2_exchange(request.GET)
http = httplib2.Http()
http = credential.authorize(http)
service = build("plus", "v1", http=http)
people_resource = service.people()
people_document = people_resource.get(userId='me').execute()
email = people_document['emails'][0]['value']
firstname = people_document['name']['givenName']
lastname = people_document['name']['familyName']
fullname = "{} {}".format(firstname, lastname)
try:
user = User.objects.get(email=email)
except User.DoesNotExist:
user = AuthHelper().save_user_profile(email,
firstname, lastname)
AuthHelper().save_user_mahasiswa_profile(user,
None, fullname, None)
# make user is authenticated
auth_login(request, user)
redirect_uri = os.getenv('FE_URL_V1', '') + "mahasiswa"
response = HttpResponseRedirect(redirect_uri)
response.set_cookie("user_id", user.id)
response.set_cookie("username", user.username)
response = JWTHelper().store_to_cookies(response, user, 3)
return response
