Django로 사내 업무 페이지 만들기-1

이전에 만들던 놈은 야근이 너무 잦아져서 하도 못건들다가 코드를 잊어먹는 바람에 드랍됐다.

그리고 나를 업무 과중으로 몰고가던 업무가 하나 있었는데 그걸 땜빵하고자, 드랍됐던 홈페이지를 마저 만들다가 이 업무를 다른 사람에게 주게 된 것도 한목해서 어영부영 드랍됐다.(행복한 드랍)

 

그치만, 이제 다시 부활했다.

 

---

"너가 이전에 한 번 해봤으니까 또 다시 한 번 해주라."

"아.. 차장님.. 다른건 괜찮은데 그거하다가 저 기절할 뻔 했었어요."

"그럼 나보고 기절하란 말이니?"

"안입니다"

---

 

그 업무의 내용인즉, 재고를 관리하기 위해 수불 관리를 하는 것인데,

이놈의 회사는 석기시대 마냥 엑셀로 전부 기입해 그걸 통계내고 틀린거 찾아내고 비어진 물량을 찾아내고 그런다.

 엑셀이라는 것이, 파일 하나를 혼자쓰고 정리할 때는 참 괜찮은건 맞다. 그치만 각자 다른 양식을 써서 같은 결과를 도출하고자 한다면 이거만큼 비효율적인게 없다.

 10 명이 관리를 하면, 최대 10 개 만큼의 양식이 나온다. 이거 다챙겨보는 사람은 진짜 매의눈 그자체다.

 

1. 업무 플로우 확인 및 모델 재고.

• 각자 관리하고 있는 재고는 그냥 자기가 알아서 입력한다.
• 해당 재고에 관해서는 타 관리자가 조회는 할 수 있지만, 각자 독립된 결과를 산출할 수 있도록 한다.
• 이 독립된 결과들을 확인해 누락되는 결과가 없는지를 수시로 확인할 수 있게끔 한다.
• 엑셀로 다운받아야한다.(무조건. 아 그놈의 엑셀)
• 즉, 관리자 갑이 A->B로 5 건을 입력했다면, 관리자 을 또한 B->A로 5건을 입력해야 하며, 이 재고를 서로 크로스 체크하기 위함이다.

 

2. 이전에 땡겨온것.

모델 - 다행스럽게 이전에 짰던 모델이 있기에 그냥 그걸 복붙해서 마저 짰다.

from django.db import models
from django.utils import timezone
import datetime
from django.core.exceptions import ValidationError


class Locations(models.Model):
    location = models.CharField(
        max_length=10, 
        verbose_name="출고지"
    )

    def __str__(self):
        return self.location


class Categories(models.Model):
    category = models.CharField(
        max_length=20, 
        verbose_name="구분"
    )

    def __str__(self):
        return self.category


class SubCategories(models.Model):
    sub_category = models.CharField(
        max_length = 20, 
        verbose_name = "서포트 세부구분"
    )
    
    def __str__(self):
        return self.sub_category


class DetailedReceiptInformation(models.Model):
    category = models.ForeignKey(
        Categories, 
        on_delete = models.CASCADE, 
        help_text = "카테고리"
    )

    sub_category = models.ForeignKey(
        SubCategories, 
        null=True, 
        blank=True, 
        on_delete = models.CASCADE, 
        help_text = "서포트일 경우"
    )

    weight = models.IntegerField(
        default = 0, 
        verbose_name = "중량"
    )

    is_steel = models.BooleanField(
        default = False, 
        blank = True,
        verbose_name = "스틸유무"
    )

    is_alu = models.BooleanField(
        default = False, 
        blank = True, 
        verbose_name = "알루유무"
    )

    is_support = models.BooleanField(
        default = False, 
        blank = True,
        verbose_name = "서포트유무"
    )


class IntegratedReceiptInformation(models.Model):
    from_location = models.ForeignKey(
        Locations, 
        on_delete = models.CASCADE, 
        related_name = "from_location",
        # null = True, 
        # blank = True,
    )

    to_location = models.ForeignKey(
        Locations, 
        on_delete = models.CASCADE, 
        related_name = "to_location", 
        # null = True, 
        # blank = True,
    )

    information = models.OneToOneField(
        DetailedReceiptInformation,
        on_delete = models.CASCADE,
        # blank = True, 
        # null = True,
    )

    contain_location = models.ForeignKey(
        Locations,
        on_delete=models.CASCADE,
        default=1,
    )

    registered_date = models.DateField(
        verbose_name = "등록일", 
        # auto_now_add = True,
        default=datetime.date.today
    )
    
    modified_date = models.DateField(
        verbose_name = '수정일',
        auto_now = True,
    )

    memo = models.CharField(
        max_length = 50, 
        verbose_name = "메모", 
        blank = True, 
        null = True,
    )

    # def save(self, *args, **kwargs):
    #     if self.from_location != self.contain_location:
    #         self.contain_location = self.from_location

    #     return super(IntegratedReceiptInformation,self).save(*args, **kwargs)

기초 자료 입력 - CommandLine을 통해 자료를 입력할 수 있게끔 기초자료를 입력할 수 있도록 코드를 작성했다.

 

3. 테스트 코드부터 작성

갈 길이 막막하기에 일단 테스트코드들부터 작성했다.

아래는 post와 get을 테스트하는 코드.

물론 이것저것 진행하면서 메인코드가 변경됐기에 이거는 잘 안될 수도 있다.

import json
from django.db import transaction
from django.test import TestCase, Client
from django.contrib.auth.models import User
import json
from django.urls import reverse
# from alu_stock.models import *
# from alu_stock.API.Serializers.stock_serializer import ManagementForWeightOfStockSerializer
from django.http import QueryDict
from ..models import *
from ..API.Serializers.stock_serializer import ManagementForWeightOfStockSerializer

# c = Client()
# response = c.post('/api/api-token-auth/', {'username':'nadure', 'password':'abc123'})
# TestCase.assertFalse(True)

# Setup run before every test method.
def set_initial_data():
    '''
    기초 자료 입력
    '''
    location_list = ['보호보호']

    category_list = ['보호보호', ]

    subcategory_list = ['보호보호']

    loc_bulk_list = []
    for loc in location_list:
        loc_bulk_list.append(Locations(location=loc))
    Locations.objects.bulk_create(loc_bulk_list)

    category_bulk_list = []
    for cat in category_list:
        category_bulk_list.append(Categories(category=cat))
    Categories.objects.bulk_create(category_bulk_list)

    sub_category_bulk_list = []
    for sub in subcategory_list:
        sub_category_bulk_list.append(SubCategories(sub_category=sub))
    SubCategories.objects.bulk_create(sub_category_bulk_list)


class PostGetTest(TestCase):
    def setUp(self):
        set_initial_data()
        my_admin = User.objects.create_superuser('nadure', 'myemail@test.com', 'password')
        self.set_dummy_data()
        
    def set_dummy_data(self):
        # Save Dummy Data
        test_data = {
            'from_location':'공장',
            'to_location':'보호',
            'category':'원자재',
            'weight':70,
            'is_alu':True,
            'is_support':False,
            'memo':'test'
        }
        from_location, to_location, category, sub_category = '공장', '보호', '원자재', None
        weight, is_alu, is_support, memo = 70, True, False, 'test'

        with transaction.atomic():
            try:
                category = Categories.objects.get(category=category)
            except Categories.DoesNotExist:
                raise Exception('there is no category matched.')

            detailed = DetailedReceiptInformation(
                category=category,
                sub_category = sub_category,
                weight=weight,
                is_alu=is_alu,
                is_support=is_support,
            )
            
            detailed.save()

            location1 = Locations.objects.get(location=from_location)
            location2 = Locations.objects.get(location=to_location)

            integrated = IntegratedReceiptInformation(
                from_location=location1,
                to_location=location2,
                information_id = detailed.id,
                memo=memo
                # registered_date=registered_date
            )
            integrated.save()
            
        # returning integrated instace
        return integrated



    def get_token(self):
        # c = Client()
        response = self.client.post(reverse('api:get_token'), {'username':'nadure', 'password':'password'})
        res = json.loads(response.content)
        token = res.get('token')
        return token

    def test_auth_use_test(self):
        token = self.get_token()
        c = Client(HTTP_AUTHORIZATION='Token '+ token)
        response = c.get(reverse('api:auth_test'))
        self.assertEquals(response.status_code, 200)

    def test_post_data_test(self):
        token = self.get_token()
        c = Client(HTTP_AUTHORIZATION='Token '+ token)
        test_data = {
            'from_location':'보호보호',
            'to_location':'보호보호',
            'category':'원자재',
            'weight':70,
            'is_alu':True,
            'is_support':False,
            'memo':'test'
        }
        response = c.post(reverse('api:write_inout'), data=test_data)
        self.assertEquals(response.status_code, 201)
        
    def test_get_integrated_data_test(self):
        # 통합 정보 get 테스트

        token = self.get_token()
        c = Client(HTTP_AUTHORIZATION='Token '+ token)
        response = c.get(reverse('api:write_inout'), data={'pk':1})
        print(response.content)
        self.assertEquals(response.status_code, 200)


    def test_treating_location_data(self):
        token = self.get_token()
        c = Client(HTTP_AUTHORIZATION='Token '+ token)
        
        response = c.get(reverse('api:subinformation'), data={'about':'location'})
        self.assertEquals(response.status_code, 200)

        response = c.post(
            reverse('api:subinformation'), 
            data={'about':'location', 'content':'test'}
            )
        self.assertEquals(response.status_code, 201)


    def tearDown(self):
        # Clean up run after every test method.
        # self.assertFalse(True)
        # print('teardown')
        pass

    def test_something_that_will_pass(self):
        self.assertFalse(False)

 

4. Django DRF를 써서 REST 서버를 만드는걸로 바꿈

 아무래도 뭔가 데이터베이스를 사용하려면 서버를 통해야할 것이고, 이 데이터베이스를 잘 사용하는 방법중에 하나는 REST API를 만드는 것이다.

 홈페이지를 만들어서 이걸로 이것저것 만들자니, 일단은 시간이 오래걸렸다. 홈페이지는 암만 허접해도 데이터를 잘 보여주기위해서 쌩쑈를 해야한다. 빨리 빨리 만드는게 중요한게 아니다. 어쨌뜬 잘 보여지게 만들어야한다. 그러다보니, 디자인은 어느정도 먹고 들어가는 윈도우 프로그래밍쪽으로 눈을 돌릴 수 밖에 없어보였다.

 순전히 빨리 만들기 위해서 DRF를 사용키로 했다.

지금 어느정도 완성이 됐는데, 생각보다 빨리 만들 수 있으면서 유연성또한 괜찮았다.

 

4-1 유연성을 챙기는 방법에 대하여

 단편적인 REST API를 구축할 때는 하나의 정보를 받기 위해선 하나의 END POINT 가 있어야 한다. 즉,

http://127.0.0.1:8000/api/information/base/

이놈은 하나의 정보만을 줄 수 밖에 없다는 뜻인데, GET, PUT 등의 파라미터를 이용하면 주어진 파라미터 또는 전송받는 데이터를 통해 유연하게 대처할 수 있다.

예를 들면,

http://127.0.0.1:8000/api/information/base/?about=location

이 주소로 요청하게 된다면

self.request.GET.get('about')

에서는 Location이라는 값을 가져온다. 그러니까 이걸 통해서 분기를 줄 수 있으면서 코드를 조금 더 유연하게 대처할 수 있다는거다.

 

4-2 그래서 뭐

METHOD	역할
POST	생성
GET	조회
PUT	수정
DELETE	삭제

 각 메써드는 위와 같은 역할들을 하고 있다. 사실 뭘로 오던 처리하면 되긴 하는데, 위처럼 약속하고 코드를 짠다고 보면 좋을듯 하다.

 그러니까, 지금 나의 경우에는 어떠한 종합적인 자료들에 관해서 쿼리를 수행하고 그 결과값을 주어야한다. 이 쿼리는 무언가 결여된 쿼리 일 수도 있고, 무언가 꽉 차 있어서 하나라도 흘리면 값이 틀어질 수도 있을 그런 쿼리들이다.

코드로 예시를 보여주자면,

# 전체 조회
qs=IntegratedReceiptInformation.objects.all()


# 필터1
qs = qs.filter(information__is_steel=self.is_steel)\
	.filter(information__is_alu=self.is_alu)\
    .filter(information__is_support=self.is_support)
   

# 필터2
qs = qs.filter(information__category__category=specific_category)

 

뭔가 필터가 많아진다. 여러 경우의 수에 대해서 필터를 준비해주는게 너무 복잡해지게 된다.

 그러다보니 그냥 데이터를 모아서 분기별로 필터링을 해주는 헬퍼를 사용하는 방식을 채택했다. 보면 변수로 뷰에서 받는 request를 통째로 받는다.

class QueryHelper(object):
    def __init__(self, request, qs=IntegratedReceiptInformation.objects):
        self.date_range_from = request.GET.get('date_range_from', None)
        self.date_range_to = request.GET.get('date_range_to', None)
        self.contain_location = request.GET.get('contain_location', None)
        self.to_location = request.GET.get('to_location', None)
        self.from_location = request.GET.get('from_location', None)
        self.specific_category = request.GET.get('specific_category', None)
        self.result_sum_only = request.GET.get('result_sum_only', False)

        self.is_steel = request.GET.get('is_steel', False)
        self.is_alu = request.GET.get('is_alu', False)
        self.is_support = request.GET.get('is_support', False)

        self.qs = qs

        if self.date_range_from and self.date_range_to :
            try:
                convert_date = datetime.datetime.strptime(self.date_range_from, '%Y-%m-%d').date()
                convert_date = datetime.datetime.strptime(self.date_range_to, '%Y-%m-%d').date()
            
            except:
                raise ValueError('range string is not valid.')
            
            self.qs = self.qs.filter(registered_date__range=[
                self.date_range_from, self.date_range_to
            ])
        
        self.qs = self.qs.filter(information__is_steel=self.is_steel)\
            .filter(information__is_alu=self.is_alu)\
            .filter(information__is_support=self.is_support)

        if self.contain_location:
            self.qs = self.qs.filter(contain_location__location=self.contain_location)
        
        if self.to_location:
            self.qs = self.qs.filter(to_location__location=self.to_location)

        if self.from_location:
            self.qs = self.qs.filter(from_location__location=self.from_location)
        
        if self.specific_category:
            self.qs = self.qs.filter(information__category__category=self.specific_category)

 이렇게 filter를 많이 하더라도 그저 추가되는 쿼리 스트링을 담기만 하는 것이기 때문에 실질적인 쿼리는 하나만 발생된다.

이 방식은 진짜 편하다.

 

예를 들면 이런 비정형의 쿼리식을 하나하나 다 따로 처리하기에는 너무 힘들다.

 

---

그렇게 해서 만든 쿼리 함수

class QueryForSumByDate(object):
    # from alu_stock.API.queryset import *
    def __init__(self, contain_location_id, qs=None, date=datetime.date.today()):
        '''
        contain_location_id -> str: The field in IntegratedReceiptInformation model
        qs - > query object: from IntegratedReceiptInformation
        date -> datetime object
        '''

        if qs is None:
            t = IntegratedReceiptInformation.objects

        else:
            t = qs
        
        self.qs = t
        
        self.contain_location = Locations.objects.get(id=contain_location_id).location

        # self.qs = t.filter(contain_location__id=contain_location_id)\
        #            .filter(registered_date__year=date.year)\
        #            .filter(registered_date__month=date.month)
        
        self.result = self._annotate_by_date()
        self.result_sum_only = self.result.values(
            'day','get_total_sum', 
            'get_out_day_sum', 'get_in_day_sum', 
            )
    
    def _annotate_by_date(self):
        # temp_qs = self.qs.filter
        # .annotate(get_in_day_sum=Sum('information__weight'))\ # original
        res = self.qs.annotate(day=TruncDay('registered_date'))\
            .values('day')\
            .annotate(
                get_in_day_sum=
                Sum(
                    Case(
                        When(
                            to_location__location=self.contain_location,
                            then='information__weight'
                        ), default=Value(0)
                    )
                )
            )\
            .annotate(
                get_out_day_sum=
                Sum(
                    Case(
                        When(
                            from_location__location=self.contain_location,
                            then='information__weight'
                        ), default=Value(0)
                    )
                )
            )\
            .annotate(get_total_sum=Sum('information__weight'))\
            .annotate(category=F('information__category__category'))\
            .annotate(sub_category=F('information__sub_category__sub_category'))\
            .annotate(from_location=F('from_location__location'))\
            .annotate(to_location=F('to_location__location'))\
            .annotate(weight=F('information__weight'))\
            .annotate(is_alu=F('information__is_alu'))\
            .annotate(is_steel=F('information__is_steel'))\
            .annotate(is_support=F('information__is_support'))\
            .values(
                'is_alu', 'is_steel', 'is_support',
                'weight', 'sub_category', 'id', 'from_location', 'to_location',
                'day','get_total_sum', 'get_out_day_sum', 'get_in_day_sum', 
                'category', 'registered_date','modified_date', 'memo'
                )
        # res = t.annotate(day=TruncDay('registered_date')).values('day').annotate(day_sum=Sum('information__weight')).values('day','day_sum')
        return res

 

대충 이런식으로 쿼리를 짜주고, 윈도우 프로그램으로 넘어갔다. 아직 해야할 거는 세부 쿼리들 정도..?

참고로 TruncDay('regisered_date')는 일자별 데이터를 뽑을 수 있게끔 도와준다.

 

 

5. 로그인 폼

PyQt5로 대충 만들어봤다.

 조금 간단한 방식으로 디자인을 했다. 아 이래서 REST를 쓰는구나 싶을 정도로 서버에서 주는 값만을 받으니 UI에서는 값 땡겨오는것만 처리하면 되기에 너무 코딩하기 편했다.

 그냥 이뻐서 먼저 가져와 봤다.

 

메인클라이언트는

아직 만드는 중이다.

 애초에 아무런 기획 자체가 없었기 때문에 내맘대로 만드는 중이라, 언제 뭐가 바뀌고 어떻게 추가될 지 몰라서 그냥 디자인은 대충대충 만들고 있다.

 

6. 시리얼라이저

 아무래도 모델 시리얼라이저 등 이미 구현돼 있는 시리얼라이저를 사용하게 된다면, 간단한 형태의 경우엔 괜찮지만 복잡한 형태의 경우에는 상당히 난항을 겪게 된다. 예를 들면 외래키 사용에서 힘든데, 모든 참조되는 모델들에 대해서 대응하는 만큼 시리얼라이저를 만들어야 하기에 코드의 가독성이 나빠지는 우려가 있었다. 그래서 그냥 쌩으로 시리얼라이저 하나를 크게 만들기로 결정했다. 이게 더 보기도 편하고 쓰기도 편하긴 하다.

class ManagementForWeightOfStockSerializer(serializers.Serializer):
    """
    재고중량 데이터 저장에 관한 시리얼라이저.
    """
    
    from_location = serializers.CharField(max_length=10)
    to_location = serializers.CharField(max_length=10)
    contain_location = serializers.CharField(max_length=10)
    category = serializers.CharField(max_length=20)
    sub_category = serializers.CharField(max_length=20, allow_null=True, allow_blank=True, required=False)

    weight = serializers.IntegerField()
    is_alu = serializers.BooleanField()
    is_support = serializers.BooleanField()

    memo = serializers.CharField(max_length=50, required=False)

    class Meta:
        validators = []

    def create(self, validated_data):
        print(validated_data)
        # 변수 취합
        from_location = validated_data.get('from_location')
        to_location = validated_data.get('to_location')
        contain_location = validated_data.get('contain_location')
        category = validated_data.get('category')
        sub_category = validated_data.get('sub_category', None)
        weight = validated_data.get('weight')

        is_alu = validated_data.get('is_alu', False)
        is_steel = validated_data.get('is_steel', False)
        is_support = validated_data.get('is_support', False)

        memo = validated_data.get('memo')
        registered_date = validated_data.get('registered_date', datetime.date.today().strftime('%Y-%m-%d'))
        registered_date = datetime.datetime.strptime(registered_date, '%Y-%m-%d').date()
        
        # 원자성 확보
        with transaction.atomic():
            try:
                category = Categories.objects.get(category=category)
            except Categories.DoesNotExist:
                raise Exception('there is no category matched.')

            detailed = DetailedReceiptInformation(
                category=category,
                sub_category = sub_category,
                weight=weight,
                is_alu=is_alu,
                is_support=is_support,
                is_steel =is_steel,
            )
            
            detailed.save()

            location1 = Locations.objects.get(location=from_location)
            location2 = Locations.objects.get(location=to_location)
            location3 = Locations.objects.get(location=contain_location)

            integrated = IntegratedReceiptInformation(
                from_location_id=location1.id,
                to_location_id=location2.id,
                contain_location_id=location3.id,
                information_id = detailed.id,
                memo=memo,
                registered_date=registered_date
            )
            integrated.save()
        
        
        # returning integrated instace
        return integrated

 뭔가 큼지막한거 같지만, 실상 모델시리얼라이저로 구성한다 치더라도 이거만큼은 적어야 하지 싶다.

 

이걸 사용하는 뷰는 이렇다.

class ManagementForWeightOfStock(APIView):
    """
    ManagementForWeightStock

    * Requires token authentication.
    * Only admin users are able to access this view.

    """
    authentication_classes = [authentication.TokenAuthentication]
    permission_classes = [permissions.IsAdminUser]

    def get(self, request, format=None):
        """
        Return a list of all users.
        """
        # usernames = [user.username for user in User.objects.all()]
        # print(IntegratedReceiptInformation.objects.all())

        pk = request.GET.get('pk')

        obj = IntegratedReceiptInformation.objects.get(id=pk)
        res = model_to_dict(obj)
        res['from_location'] = Locations.objects.get(id=res['from_location']).location
        res['to_location'] = Locations.objects.get(id=res['to_location']).location
        res['information'] = model_to_dict(DetailedReceiptInformation.objects.get(id=res['information']))
        return Response(res, status=status.HTTP_200_OK)

    def post(self, request, *args, **kwargs):
        """
        Make Integrated data for managing weight.
        """

        serializer = ManagementForWeightOfStockSerializer(data=request.data)
        # print(serializer.data)

        if serializer.is_valid(raise_exception=True):
            serializer.save()
            # print(request.data)
            # print(serializer.data)
            return Response(request.data, status=status.HTTP_201_CREATED)

        else:
            # print(request.data)
            # print(serializer.errors)
            return Response(serializer.errors, status=status.HTTP_400_BAD_REQUEST)

 시리얼라이저에 request.data를 넣는다. 이부분에서 상당히 헷갈리는것이 어떨 때는 many=True를 써줘야 하지만 지금은 안써주니까 잘 들어간다. 뭐가 맞는진 모르겠으나, 지금은 이게 맞다. 받는 데이터의 형식에 따라 달라지는것 같다. type( request.data.get('some_field') ) 이 리스트일 때 many=True를 받는것 같다.

 

7. 헬퍼 클래스는 인정

 데이터를 받아다 정리를 하는 부분들을 헬퍼클래스로 몰아다 놓고 여기에서 하부 클래스로 뽑아다 쓰니 코드 정리도 되고 여러모로 좋았다. 여기에다 각종 예외들을 추가하기도 쉬울듯 하다.

 

8. 잘 모르겠으면 일단 테스트 코드부터.

 테스트 코드는 2 가지 느낌으로 사용했다.

1. 코드를 어떻게 짜나가야 할 지 막막할 때
2. 내가 사용한 코드가 잘 되는지 이것저것 테스트할 때

 1번에서 계속해서 테스트 코드를 돌려가면서 첫 시작 코드를 적는다. 그렇게 해서 만들어진 코드를 참고해 만들고자 했던 코드로 베껴서 다시 정리하며 작성한다.

 그렇게 1번을 통해 만들어진 사용할 코드를 임포트해서 테스트해본다. 이게 리얼 테스트코드.

 

9. TODO

Authentication - 회원가입, 비밀번호 변경등을 위해사용할 예정이다. 비밀번호 변경이 귀찮으면 아이디 삭제했다가 다시 만드는걸로 해야겠다. 할 줄 모른다고 잡아떼면 그만

대쉬보드 형 쿼리 - 대쉬보드에서 사용할 쿼리를 땡긴다. 어떠한 형태로 쿼리를 원하는지가 전혀 정해지지 않았기 때문에, 일단은 대충 내맘대로 만들어야 한다.

Excel Export - 아 진짜. 고놈의 엑셀. 엑셀이 끼게 되면 일단 서버의 성능이 조금 더 올라가야한다. 이 부분 때문에 어쩔 수 없이 웹이 아닌 프로그램의 형태를 띄게 됐다. 데이터는 다 줄테니, 알아서 엑셀 짜서 해!

 

끝.