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
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
| import sqlite3
from typing import Any, Dict, List
class DatabaseField:
"""데이터베이스 필드 정의"""
def __init__(self, field_type, primary_key=False, nullable=True, default=None):
self.field_type = field_type
self.primary_key = primary_key
self.nullable = nullable
self.default = default
def to_sql_type(self):
"""Python 타입을 SQL 타입으로 변환"""
type_mapping = {
int: 'INTEGER',
str: 'TEXT',
float: 'REAL',
bool: 'BOOLEAN'
}
return type_mapping.get(self.field_type, 'TEXT')
class QuerySet:
"""쿼리 실행을 위한 클래스"""
def __init__(self, model_class, db_connection):
self.model_class = model_class
self.db_connection = db_connection
def create(self, **kwargs):
"""새 레코드 생성"""
fields = list(self.model_class._fields.keys())
placeholders = ', '.join(['?' for _ in fields])
field_names = ', '.join(fields)
values = [kwargs.get(field, self.model_class._fields[field].default)
for field in fields]
query = f"INSERT INTO {self.model_class._table_name} ({field_names}) VALUES ({placeholders})"
cursor = self.db_connection.execute(query, values)
self.db_connection.commit()
# 생성된 객체 반환
kwargs['id'] = cursor.lastrowid
return self.model_class(**kwargs)
def all(self):
"""모든 레코드 조회"""
query = f"SELECT * FROM {self.model_class._table_name}"
cursor = self.db_connection.execute(query)
results = []
for row in cursor.fetchall():
field_names = list(self.model_class._fields.keys())
kwargs = dict(zip(field_names, row))
results.append(self.model_class(**kwargs))
return results
def filter(self, **kwargs):
"""조건에 맞는 레코드 필터링"""
conditions = []
values = []
for field, value in kwargs.items():
conditions.append(f"{field} = ?")
values.append(value)
where_clause = " AND ".join(conditions)
query = f"SELECT * FROM {self.model_class._table_name} WHERE {where_clause}"
cursor = self.db_connection.execute(query, values)
results = []
for row in cursor.fetchall():
field_names = list(self.model_class._fields.keys())
obj_kwargs = dict(zip(field_names, row))
results.append(self.model_class(**obj_kwargs))
return results
class ORMMeta(type):
"""ORM을 위한 메타클래스"""
def __new__(cls, name, bases, attrs):
# DatabaseField 인스턴스들을 찾아서 처리
fields = {}
for key, value in list(attrs.items()):
if isinstance(value, DatabaseField):
fields[key] = value
attrs.pop(key) # 필드는 클래스 속성에서 제거
# 메타데이터 설정
attrs['_fields'] = fields
attrs['_table_name'] = name.lower()
new_class = super().__new__(cls, name, bases, attrs)
# 동적으로 프로퍼티 생성
for field_name, field in fields.items():
cls._create_property(new_class, field_name, field)
return new_class
@staticmethod
def _create_property(cls, field_name, field):
"""필드에 대한 프로퍼티 생성"""
private_name = f'_{field_name}'
def getter(self):
return getattr(self, private_name, field.default)
def setter(self, value):
if not field.nullable and value is None:
raise ValueError(f"{field_name} cannot be None")
if value is not None and not isinstance(value, field.field_type):
# 타입 변환 시도
try:
value = field.field_type(value)
except (ValueError, TypeError):
raise TypeError(f"{field_name} must be of type {field.field_type.__name__}")
setattr(self, private_name, value)
setattr(cls, field_name, property(getter, setter))
class Model(metaclass=ORMMeta):
"""ORM 모델 기본 클래스"""
_db_connection = None
@classmethod
def set_database(cls, db_path):
"""데이터베이스 연결 설정"""
cls._db_connection = sqlite3.connect(db_path, check_same_thread=False)
cls._create_table()
@classmethod
def _create_table(cls):
"""테이블 생성"""
if not cls._fields:
return
columns = []
for field_name, field in cls._fields.items():
column_def = f"{field_name} {field.to_sql_type()}"
if field.primary_key:
column_def += " PRIMARY KEY"
if not field.nullable:
column_def += " NOT NULL"
columns.append(column_def)
columns_sql = ", ".join(columns)
query = f"CREATE TABLE IF NOT EXISTS {cls._table_name} ({columns_sql})"
cls._db_connection.execute(query)
cls._db_connection.commit()
@classmethod
def objects(cls):
"""QuerySet 반환"""
return QuerySet(cls, cls._db_connection)
def __init__(self, **kwargs):
for field_name in self._fields:
if field_name in kwargs:
setattr(self, field_name, kwargs[field_name])
def __str__(self):
field_values = []
for field_name in self._fields:
value = getattr(self, field_name)
field_values.append(f"{field_name}={value}")
return f"{self.__class__.__name__}({', '.join(field_values)})"
# 모델 정의
class User(Model):
id = DatabaseField(int, primary_key=True)
name = DatabaseField(str, nullable=False)
email = DatabaseField(str, nullable=False)
age = DatabaseField(int, default=0)
class Post(Model):
id = DatabaseField(int, primary_key=True)
title = DatabaseField(str, nullable=False)
content = DatabaseField(str, default="")
user_id = DatabaseField(int, nullable=False)
# 사용 예제
if __name__ == "__main__":
# 데이터베이스 설정
User.set_database(":memory:") # 메모리 데이터베이스 사용
Post.set_database(":memory:")
# 사용자 생성
user1 = User.objects().create(name="Alice", email="alice@example.com", age=25)
user2 = User.objects().create(name="Bob", email="bob@example.com", age=30)
print("Created users:")
for user in User.objects().all():
print(f" {user}")
# 포스트 생성
post1 = Post.objects().create(title="Hello World", content="First post", user_id=user1.id)
post2 = Post.objects().create(title="Python Tips", content="Some tips", user_id=user1.id)
print("\nAll posts:")
for post in Post.objects().all():
print(f" {post}")
# 필터링
alice_posts = Post.objects().filter(user_id=user1.id)
print(f"\nAlice's posts ({len(alice_posts)}):")
for post in alice_posts:
print(f" {post}")
|