master
commit
ec327db173
@ -0,0 +1,288 @@
|
||||
from test import Student
|
||||
import os.path
|
||||
from abc import ABC, abstractmethod
|
||||
from datetime import date, datetime
|
||||
from typing import List, Dict, Optional
|
||||
import json
|
||||
import csv
|
||||
|
||||
class IStudentDAL(ABC):
|
||||
|
||||
@abstractmethod
|
||||
def get_by_id(self, id: str) -> Optional[[Student]]:
|
||||
pass
|
||||
|
||||
@abstractmethod
|
||||
def get_by_stu_id(self, stu_id: str) -> Optional[Student]:
|
||||
pass
|
||||
|
||||
@abstractmethod
|
||||
def get_all(self) -> List[Student]:
|
||||
pass
|
||||
|
||||
@abstractmethod
|
||||
def add(self, student: Student) -> bool:
|
||||
pass
|
||||
|
||||
@abstractmethod
|
||||
def delete(self, id: str) -> bool:
|
||||
pass
|
||||
|
||||
@abstractmethod
|
||||
def delete_by_stu_id(self, stu_id: str) -> bool:
|
||||
pass
|
||||
|
||||
@abstractmethod
|
||||
def update(self, student: Student) -> bool:
|
||||
pass
|
||||
|
||||
@abstractmethod
|
||||
def is_exist(self, id: str) -> bool:
|
||||
pass
|
||||
|
||||
@abstractmethod
|
||||
def is_exist_stu_id(self, stu_id: str) -> bool:
|
||||
pass
|
||||
|
||||
@abstractmethod
|
||||
def import_from_json(self, file_path: str) -> List[Dict]:
|
||||
pass
|
||||
|
||||
@abstractmethod
|
||||
def export_to_json(self, file_path: str) -> bool:
|
||||
pass
|
||||
|
||||
@abstractmethod
|
||||
def import_from_csv(self, file_path: str) -> List[Dict]:
|
||||
pass
|
||||
|
||||
@abstractmethod
|
||||
def export_to_csv(self, file_path: str) -> bool:
|
||||
pass
|
||||
|
||||
class JsonStudentDAL(IStudentDAL):
|
||||
|
||||
def __init__(self, file_path: str):
|
||||
self.file_path = file_path
|
||||
self._ensure_file_exists()
|
||||
|
||||
def _ensure_file_exists(self):
|
||||
if not os.path.exists(self.file_path):
|
||||
with open(self.file_path, 'w', encoding='utf-8') as f:
|
||||
json.dump([], f, ensure_ascii=False)
|
||||
|
||||
def _load(self) -> List[Dict]:
|
||||
with open(self.file_path, 'r', encoding='utf-8') as f:
|
||||
return json.load(f)
|
||||
|
||||
def _save(self, data: List[Dict]):
|
||||
with open(self.file_path, 'w', encoding='utf-8') as f:
|
||||
json.dump(data, f, ensure_ascii=False, indent=2)
|
||||
|
||||
def get_by_id(self, id: str) -> Optional[Student]:
|
||||
data = self._load()
|
||||
for item in data:
|
||||
if item['id'] == id:
|
||||
return Student.from_dict(item)
|
||||
return None
|
||||
|
||||
def get_by_stu_id(self, stu_id: str) -> Optional[Student]:
|
||||
data = self._load()
|
||||
for item in data:
|
||||
if item['stu_id'] == stu_id:
|
||||
return Student.from_dict(item)
|
||||
return None
|
||||
|
||||
def get_all(self) -> List[Student]:
|
||||
return [Student.from_dict(item) for item in self._load()]
|
||||
|
||||
def add(self, student: Student) -> bool:
|
||||
data = self._load()
|
||||
if self.is_exist(student.id):
|
||||
return False
|
||||
|
||||
data.append(student.to_dict())
|
||||
self._save(data)
|
||||
return True
|
||||
|
||||
def delete(self, id: str) -> bool:
|
||||
data = self._load()
|
||||
new_data = [item for item in data if item['id'] != id]
|
||||
if len(data) == len(new_data):
|
||||
return False
|
||||
|
||||
self._save(new_data)
|
||||
return True
|
||||
|
||||
def delete_by_stu_id(self, stu_id: str) -> bool:
|
||||
data = self._load()
|
||||
new_data = [item for item in data if item['stu_id'] != stu_id]
|
||||
if len(data) == len(new_data):
|
||||
return False
|
||||
|
||||
self._save(new_data)
|
||||
return True
|
||||
|
||||
def update(self, student: Student) -> bool:
|
||||
data = self._load()
|
||||
updated = False
|
||||
|
||||
for i, item in enumerate(data):
|
||||
if item['id'] == student.id:
|
||||
data[i] = student.to_dict()
|
||||
updated = True
|
||||
break
|
||||
|
||||
if updated:
|
||||
self._save(data)
|
||||
return updated
|
||||
|
||||
def is_exist(self, id: str) -> bool:
|
||||
return any(item['id'] == id for item in self._load())
|
||||
|
||||
def is_exist_stu_id(self, stu_id: str) -> bool:
|
||||
return any(item['stu_id'] == stu_id for item in self._load())
|
||||
|
||||
def import_from_json(self, file_path: str) -> List[Dict]:
|
||||
with open(file_path, 'r', encoding='utf-8') as f:
|
||||
return json.load(f)
|
||||
|
||||
def export_to_json(self, file_path: str) -> bool:
|
||||
try:
|
||||
with open(file_path, 'w', encoding='utf-8') as f:
|
||||
json.dump(self._load(), f, ensure_ascii=False, indent=2)
|
||||
return True
|
||||
except:
|
||||
return False
|
||||
|
||||
class CsvStudentDAL(IStudentDAL):
|
||||
"""CSV格式数据访问实现类"""
|
||||
|
||||
FIELD_TYPES = {
|
||||
'id': str,
|
||||
'stu_id': str,
|
||||
'name': str,
|
||||
'age': int,
|
||||
'gender': str,
|
||||
'height': float,
|
||||
'weight': float
|
||||
}
|
||||
|
||||
def __init__(self, file_path: str):
|
||||
self.file_path = file_path
|
||||
self._ensure_file_exists()
|
||||
|
||||
def _ensure_file_exists(self):
|
||||
"""确保CSV文件存在并包含表头"""
|
||||
if not os.path.exists(self.file_path):
|
||||
with open(self.file_path, 'w', encoding='utf-8-sig', newline='') as f:
|
||||
writer = csv.DictWriter(f, fieldnames=self.FIELD_TYPES.keys())
|
||||
writer.writeheader()
|
||||
|
||||
def _convert_row(self, row: Dict) -> Dict:
|
||||
"""转换CSV行的数据类型"""
|
||||
converted = {}
|
||||
for field, dtype in self.FIELD_TYPES.items():
|
||||
value = row.get(field)
|
||||
if value is None or value == '':
|
||||
converted[field] = None
|
||||
else:
|
||||
try:
|
||||
converted[field] = dtype(value)
|
||||
except (ValueError, TypeError):
|
||||
converted[field] = None
|
||||
return converted
|
||||
|
||||
def _load(self) -> List[Dict]:
|
||||
"""从CSV文件加载数据"""
|
||||
with open(self.file_path, 'r', encoding='utf-8-sig') as f:
|
||||
reader = csv.DictReader(f)
|
||||
return [self._convert_row(row) for row in reader]
|
||||
|
||||
def _save(self, data: List[Dict]):
|
||||
"""保存数据到CSV文件"""
|
||||
with open(self.file_path, 'w', encoding='utf-8-sig', newline='') as f:
|
||||
writer = csv.DictWriter(f, fieldnames=self.FIELD_TYPES.keys())
|
||||
writer.writeheader()
|
||||
writer.writerows(data)
|
||||
|
||||
def get_by_id(self, id: str) -> Optional[Student]:
|
||||
data = self._load()
|
||||
for item in data:
|
||||
if item['id'] == id:
|
||||
return Student.from_dict(item)
|
||||
return None
|
||||
|
||||
def get_by_stu_id(self, stu_id: str) -> Optional[Student]:
|
||||
data = self._load()
|
||||
for item in data:
|
||||
if item['stu_id'] == stu_id:
|
||||
return Student.from_dict(item)
|
||||
return None
|
||||
|
||||
def get_all(self) -> List[Student]:
|
||||
return [Student.from_dict(item) for item in self._load()]
|
||||
|
||||
def add(self, student: Student) -> bool:
|
||||
data = self._load()
|
||||
if self.is_exist(student.id):
|
||||
return False
|
||||
|
||||
data.append(student.to_dict())
|
||||
self._save(data)
|
||||
return True
|
||||
|
||||
def delete(self, id: str) -> bool:
|
||||
data = self._load()
|
||||
new_data = [item for item in data if item['id'] != id]
|
||||
if len(data) == len(new_data):
|
||||
return False
|
||||
|
||||
self._save(new_data)
|
||||
return True
|
||||
|
||||
def delete_by_stu_id(self, stu_id: str) -> bool:
|
||||
data = self._load()
|
||||
new_data = [item for item in data if item['stu_id'] != stu_id]
|
||||
if len(data) == len(new_data):
|
||||
return False
|
||||
|
||||
self._save(new_data)
|
||||
return True
|
||||
|
||||
def update(self, student: Student) -> bool:
|
||||
data = self._load()
|
||||
updated = False
|
||||
|
||||
for i, item in enumerate(data):
|
||||
if item['id'] == student.id:
|
||||
data[i] = student.to_dict()
|
||||
updated = True
|
||||
break
|
||||
|
||||
if updated:
|
||||
self._save(data)
|
||||
return updated
|
||||
|
||||
def is_exist(self, id: str) -> bool:
|
||||
return any(item['id'] == id for item in self._load())
|
||||
|
||||
def is_exist_stu_id(self, stu_id: str) -> bool:
|
||||
return any(item['stu_id'] == stu_id for item in self._load())
|
||||
|
||||
def import_from_csv(self, file_path: str) -> List[Dict]:
|
||||
with open(file_path, 'r', encoding='utf-8-sig') as f:
|
||||
reader = csv.DictReader(f)
|
||||
return [self._convert_row(row) for row in reader]
|
||||
|
||||
def export_to_csv(self, file_path: str) -> bool:
|
||||
try:
|
||||
data = self._load()
|
||||
with open(file_path, 'w', encoding='utf-8-sig', newline='') as f:
|
||||
writer = csv.DictWriter(f, fieldnames=self.FIELD_TYPES.keys())
|
||||
writer.writeheader()
|
||||
writer.writerows(data)
|
||||
return True
|
||||
except:
|
||||
return False
|
||||
|
||||
Loading…
Reference in new issue