You can not select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
60 lines
2.4 KiB
60 lines
2.4 KiB
import csv
|
|
from typing import List, Optional
|
|
from pathlib import Path
|
|
from datetime import date
|
|
from src.dal.abstract_dal import IStudentDAL
|
|
from src.models.student import Student
|
|
|
|
|
|
class CSVStudentDAL(IStudentDAL):
|
|
def __init__(self, file_path: str = "data/students.csv"):
|
|
self.file_path = Path(file_path)
|
|
self.file_path.parent.mkdir(parents=True, exist_ok=True)
|
|
if not self.file_path.exists():
|
|
with open(self.file_path, "w", newline="", encoding="utf-8") as f:
|
|
writer = csv.writer(f)
|
|
writer.writerow([
|
|
"name", "id_card", "stu_id", "gender",
|
|
"height", "weight", "enrollment_date",
|
|
"class_name", "major"
|
|
])
|
|
|
|
def _read_all(self) -> List[Student]:
|
|
students = []
|
|
with open(self.file_path, "r", newline="", encoding="utf-8") as f:
|
|
reader = csv.DictReader(f)
|
|
for row in reader:
|
|
try:
|
|
student = Student(
|
|
name=row["name"],
|
|
id_card=row["id_card"],
|
|
stu_id=row["stu_id"],
|
|
gender=bool(int(row["gender"])) if row["gender"] else None,
|
|
height=int(row["height"]) if row["height"] else None,
|
|
weight=float(row["weight"]) if row["weight"] else None,
|
|
enrollment_date=row["enrollment_date"],
|
|
class_name=row["class_name"],
|
|
major=row["major"]
|
|
)
|
|
students.append(student)
|
|
except (ValueError, KeyError) as e:
|
|
continue
|
|
return students
|
|
|
|
def _write_all(self, students: List[Student]) -> bool:
|
|
try:
|
|
with open(self.file_path, "w", newline="", encoding="utf-8") as f:
|
|
writer = csv.DictWriter(f, fieldnames=[
|
|
"name", "id_card", "stu_id", "gender",
|
|
"height", "weight", "enrollment_date",
|
|
"class_name", "major"
|
|
])
|
|
writer.writeheader()
|
|
for student in students:
|
|
writer.writerow(student.to_dict())
|
|
return True
|
|
except Exception:
|
|
return False
|
|
|
|
# 实现所有抽象方法...
|
|
# 完整实现会根据上述_reader_all和_write_all方法完成各个接口 |