본문 바로가기
Back-end/Python

[Python] Data Model 만들기 (3) - Data Model

by 사용자 허도치 2021. 1. 20.
728x90

2021/01/19 - [Back-end/Python] - [Python] Data Model 만들기 (1) - BaseField

2021/01/19 - [Back-end/Python] - [Python] Data Model 만들기 (2) - Data Type Field

 

 

  이전 포스트에서 StringField, IntegerField, DatetimeField를 구현하였다. 이 Data Type Field 들의 유효성검사를 좀 더 보완해서 사용하면 더 좋겠지만, 우선은 Data Schema를 먼저 만들어 볼 계획이다. Schema는 Database에서 자료의 구조를 나타내는데, 이를 모방하여 Python Object로 구현해보려고 한다.

 

 

 

1. 에러 핸들러
1-1. ValidateError
1
2
3
4
5
6
7
8
9
10
11
12
13
14
# errors.py
 
class ValidateError(Exception):
  def __init__(self, messages, code=400):
    super(ValidateError, self).__init__()
    self.messages = messages
    self.code = code
    
  def getMessages(self):
    return self.messages
  
  def __str__(self):
    return '[{code}] {messages}'.format(code=self.code, messages=self.messages)
 
cs

 

 

 

2. Model 객체
2-1. BaseModel 객체 생성
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
# models.py
 
from fields import BaseField, StringField, IntegerField, DatetimeField
from errors import ValidateError
 
 
class BaseModel(object):
 
  def __init__(self, data=None):
    if data is not None:
      self.dump(data)
  
  def getField(self, field_name, field_type=None):
    '''
    특정 필드 조회
    * BaseField 타입의 필드를 조회하기위한 함수
    '''
    if hasattr(self, field_name):
      attr = getattr(self, field_name)
      if field_type is None:
        return attr
      elif isinstance(attr, field_type):
        return attr
      
      raise TypeError("Expected data type '%s', but '%s'." %( field_type.__name__, attr.__class__.__name__ ))
    raise ValueError("'%s' is not defined field." % ( field_name ))
 
  def getFields(self):
    '''
    필드 목록 조회
    * BaseField 타입의 필드 목록을 조회하기위한 함수
    '''
    fields = list()
    for field_name in dir(self):
      try:
        field_value = self.getField(field_name, BaseField)
        if not callable(field_value) and not field_name.startswith("_"):
          fields.append((field_name, field_value))
      except:
        pass
    return fields
  
  def __mapp(self, datas, validate=True):
    '''
    데이터 매핑
    * 각 필드에 데이터를 매핑하는 함수
    * 옵션에 따라 유효성검사를 실시
    '''
    data = dict()
    errors = dict()
 
    # 입력값 매핑
    for field_name, value in datas.items():
      try:
        field_class = self.getField(field_name, BaseField)
        field_class.setValue(value, validate=validate)
        data[field_name] = field_class.getValue()
        
      except Exception as e:
        errors[field_name] = str(e)
        
    # 기본값 매핑
    for field_name, field_class in self.getFields():
      # 입력값 매핑을 시도한 필드는 제외
      if field_name in datas:
        continue
      
      try:
        default = field_class.getOption("default"None)
        
        # 기본값이 문자열이면 현재 객체에 정의된 메소드 중 이름이 일치하는 메소드 조회
        if isinstance(default, strand hasattr(self, default):
          default = getattr(self, default)
          
        # 기본값이 메소드면 실행한 결과값을 저장하고 아니면 그냥 저장
        value = default if not callable(default) else default(field_name)
        
        # 필드에 저장된 데이터 
        field_class.setValue(value, validate=validate)
        data[field_name] = field_class.getValue()
        
      except Exception as e:
        errors[field_name] = str(e)
    
    # 값매핑 결과 반환
    return ( data, errors )
  
  def dump(self, data=None):
    '''
    단일 데이터 Dumping
    * 입력받은 데이터를 각 필드에 매핑
    * @data:Optional[Dict]
      - data가 None이면 dump 데이터로 처리(dump가 먼저 실행되어야함)
    '''
    __data = data if data is not None else self.__dump_data
    dumped, errors = self.__mapp(__data, validate=False)
    
    self.__dump_data = dumped
    self.__errors = errors
    
    return dumped
 
  def load(self, data=None):
    '''
    단일 데이터 Loading
    * 입력받은 데이터를 각 필드에 매핑
    * 유효성검사 실시, 유효하지 않은 필드는 ValidateError를 발생시켜 반환
    * @data:Optional[Dict]
      - data가 None이면 dump 데이터로 처리(dump가 먼저 실행되어야함)
    '''
    __data = data if data is not None else self.__dump_data
      
    loaded, errors = self.__mapp(__data, validate=True)
    
    self.__load_data = loaded
    self.__errors = errors
    
    if len(errors) > 0:
      raise ValidateError(errors)
      
    return loaded
    
  def __str__(self):
    return str(self.__data)
    
  def __repr__(self):
    return "<models.{self.__class__.__name__}>".format(self=self)
 
cs

 

 

2-2. UserModel 객체 생성
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
# models.py
'''
    BaseModel 밑에 이어서 작성
'''
from uuid import uuid4
from datetime import datetime
 
 
class UserModel(BaseModel):
  id = StringField(required=True, maxlength=50, default=lambda info: str(uuid4()))
  name = StringField(maxlength=10)
  role = StringField(required=True, default="user")
  age = IntegerField(required=True, min=0, max=200)
  hire_date = DatetimeField(required=Trueformat="%Y%m%d%H%M%S", default="getNowDate")
 
  def getNowDate(self, name):
    '''
        현재 시간을 반환
    '''
    return datetime.now()
 
cs

 

 

2-3. UserModel 객체 테스트
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
# test.models.py
from pprint import pprint
 
from models import UserModel
from errors import ValidateError
 
class TestUserModel(object):
  def __init__(self):
    print("UserModel 테스트")
    print("="*40)
    for attrname in dir(self):
      testcase = getattr(self, attrname)
      if attrname.startswith("test"and callable(testcase):
        print("="*40)
        print( testcase.__doc__ )
        print("="*40)
        try:
          testcase()
        except Exception as e:
          print("Error: "+str(e))
        print("")
        print("="*40)
        
  def test_case_1(self):
    '''
    데이터 매핑 방법
    '''
    user = UserModel(dict(
      name="Dumpping_1",
      age=10,
      hire_date="20210120"
    ))
    data = user.dump(dict(
      id="admin",
      name="Dumpping_2",
      age=100,
      hire_date="20210120"
    ))
        
  def test_case_2(self):
    '''
    데이터 단일 Dumping 정상 / 단일 Loading 오류
    '''
    user = UserModel(dict(
      name="Dochi",
      age=1004,
      hire_date="20210120"
    ))
    print("==== Dump Result")
    dump_data = user.dump()
    pprint( dump_data, indent=2 )
    print("")
    
    try:
      print("==== Load Result")
      load_data = user.load()
      pprint( load_data, indent=2 )
    
    except ValidateError as e:
      pprint( e.getMessages(), indent=2 )
      
  def test_case_3(self):
    '''
    데이터 단일 Dumping 정상 / 단일 Loading 정상
    '''
    user = UserModel(dict(
      name="Dochi",
      age=109,
      hire_date="20210120123456"
    ))
    print("==== Dump Result")
    dump_data = user.dump()
    pprint( dump_data, indent=2 )
    print("")
    
    try:
      print("==== Load Result")
      load_data = user.load()
      pprint( load_data, indent=2 )
    
    except ValidateError as e:
      pprint( e.getMessages(), indent=2 )
 
# 테스트실행
TestUserModel()
cs

 

 

2-4. UserModel 객체 테스트 결과

 

 

 

마치며

  - BaseModel에서 __mapp 함수가 조금 복잡하게 보일 수 있지만, 단순하게 설명하자면 입력받은 데이터를 UserModel에 정의된 필드(id, name, age, role, hire_date)에 매핑을 하는 것이다. 또한, dump와 load로 구분한 이유는 dump는 단순히 구조화된 Object에 데이터를 저장하기 위한 용도이며, load는 데이터 저장과 더불어 Database에 저장하는 등 전처리 작업에 필요한 유효성검사를 처리하는 용도이다.

  - 이번 예제에서는 단순하게 UserModel만 구현하였는데, 게시판 글을 저장하는 BoardModel이나 검색조건을 저장하는 SearchModel 등 다양하게 활용이 가능하며, 나중에는 Database Connector를 붙여서 ORM을 구현할 수도 있다.

  - 다만, 이번에 만든 BaseModel 객체는 단일 데이터만 저장할 수 있기 때문에 리스트형 데이터를 저장할 수 있는 객체를 추가로 만들거나 이번에 만든 BaseModel을 확장할 필요가 있다.

 

728x90

태그

, ,

댓글0