本文目录导读:
我来给你展示一个Python简易架构案例,用一个图书管理系统来说明分层架构的实现。
项目结构
library_system/
├── main.py # 入口文件
├── config.py # 配置模块
├── models/ # 数据模型层
│ └── book.py
├── services/ # 业务逻辑层
│ └── book_service.py
├── repositories/ # 数据访问层
│ └── book_repository.py
└── utils/ # 工具模块
└── validators.py
代码实现
配置模块 config.py
# 配置文件
class Config:
DATABASE = {
'type': 'json', # 使用JSON文件作为存储
'file_path': 'books.json'
}
数据模型层 models/book.py
from dataclasses import dataclass
from typing import Optional
import json
@dataclass
class Book:
id: Optional[int] = None str = ""
author: str = ""
isbn: str = ""
year: int = 2023
price: float = 0.0
available: bool = True
def to_dict(self):
return {
'id': self.id,
'title': self.title,
'author': self.author,
'isbn': self.isbn,
'year': self.year,
'price': self.price,
'available': self.available
}
@staticmethod
def from_dict(data):
return Book(
id=data.get('id'),
title=data.get('title'),
author=data.get('author'),
isbn=data.get('isbn'),
year=data.get('year'),
price=data.get('price'),
available=data.get('available', True)
)
数据访问层 repositories/book_repository.py
import json
import os
from typing import List, Optional
from models.book import Book
from config import Config
class BookRepository:
def __init__(self):
self.file_path = Config.DATABASE['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)
def _load_all(self) -> List[dict]:
"""加载所有数据"""
try:
with open(self.file_path, 'r', encoding='utf-8') as f:
return json.load(f)
except (json.JSONDecodeError, FileNotFoundError):
return []
def _save_all(self, books: List[dict]):
"""保存所有数据"""
with open(self.file_path, 'w', encoding='utf-8') as f:
json.dump(books, f, ensure_ascii=False, indent=2)
def get_all(self) -> List[Book]:
"""获取所有图书"""
books_data = self._load_all()
return [Book.from_dict(book) for book in books_data]
def get_by_id(self, book_id: int) -> Optional[Book]:
"""根据ID获取图书"""
books_data = self._load_all()
for book_data in books_data:
if book_data['id'] == book_id:
return Book.from_dict(book_data)
return None
def create(self, book: Book) -> Book:
"""创建新图书"""
books_data = self._load_all()
# 生成新ID
if books_data:
new_id = max(b['id'] for b in books_data) + 1
else:
new_id = 1
book.id = new_id
books_data.append(book.to_dict())
self._save_all(books_data)
return book
def update(self, book: Book) -> Optional[Book]:
"""更新图书信息"""
books_data = self._load_all()
for i, existing_book in enumerate(books_data):
if existing_book['id'] == book.id:
books_data[i] = book.to_dict()
self._save_all(books_data)
return book
return None
def delete(self, book_id: int) -> bool:
"""删除图书"""
books_data = self._load_all()
initial_length = len(books_data)
books_data = [b for b in books_data if b['id'] != book_id]
if len(books_data) < initial_length:
self._save_all(books_data)
return True
return False
def search(self, keyword: str) -> List[Book]:
"""搜索图书"""
books_data = self._load_all()
keyword = keyword.lower()
results = []
for book_data in books_data:
book = Book.from_dict(book_data)
if (keyword in book.title.lower() or
keyword in book.author.lower() or
keyword in book.isbn.lower()):
results.append(book)
return results
业务逻辑层 services/book_service.py
from typing import List, Optional
from models.book import Book
from repositories.book_repository import BookRepository
from utils.validators import BookValidator
class BookService:
def __init__(self):
self.repository = BookRepository()
def get_all_books(self) -> List[Book]:
"""获取所有图书"""
return self.repository.get_all()
def get_book(self, book_id: int) -> Optional[Book]:
"""获取单本图书"""
if book_id <= 0:
raise ValueError("图书ID必须为正整数")
return self.repository.get_by_id(book_id)
def create_book(self, title: str, author: str, isbn: str,
year: int, price: float) -> Book:
"""创建新图书"""
# 验证数据
validator = BookValidator()
errors = validator.validate(title, author, isbn, year, price)
if errors:
raise ValueError(f"数据验证失败: {', '.join(errors)}")
# 检查ISBN是否重复
existing = self.repository.search(isbn)
if any(b.isbn == isbn for b in existing):
raise ValueError(f"ISBN '{isbn}' 已存在")
book = Book(
title=title,
author=author,
isbn=isbn,
year=year,
price=price,
available=True
)
return self.repository.create(book)
def update_book(self, book_id: int, title: str = None, author: str = None,
year: int = None, price: float = None,
available: bool = None) -> Optional[Book]:
"""更新图书信息"""
existing = self.repository.get_by_id(book_id)
if not existing:
raise ValueError(f"未找到ID为 {book_id} 的图书")
# 更新字段
if title is not None:
existing.title = title
if author is not None:
existing.author = author
if year is not None:
existing.year = year
if price is not None:
existing.price = price
if available is not None:
existing.available = available
return self.repository.update(existing)
def delete_book(self, book_id: int) -> bool:
"""删除图书"""
if book_id <= 0:
raise ValueError("图书ID必须为正整数")
return self.repository.delete(book_id)
def search_books(self, keyword: str) -> List[Book]:
"""搜索图书"""
if not keyword or not keyword.strip():
return self.get_all_books()
return self.repository.search(keyword.strip())
工具模块 utils/validators.py
from typing import List
class BookValidator:
@staticmethod
def validate(title: str, author: str, isbn: str, year: int, price: float) -> List[str]:
"""验证图书数据"""
errors = []
if not title or not title.strip():
errors.append("书名不能为空")
if not author or not author.strip():
errors.append("作者不能为空")
if not isbn or not isbn.strip():
errors.append("ISBN不能为空")
elif not BookValidator._is_valid_isbn(isbn):
errors.append("ISBN格式无效")
current_year = 2024
if not (1000 <= year <= current_year):
errors.append(f"年份必须在1000到{current_year}之间")
if price < 0:
errors.append("价格不能为负数")
elif price > 10000:
errors.append("价格不能超过10000")
return errors
@staticmethod
def _is_valid_isbn(isbn: str) -> bool:
"""简单ISBN格式验证"""
isbn_clean = isbn.replace('-', '').replace(' ', '')
if len(isbn_clean) == 10:
return isbn_clean[:-1].isdigit() and (isbn_clean[-1].isdigit() or isbn_clean[-1] == 'X')
elif len(isbn_clean) == 13:
return isbn_clean.isdigit()
return False
主入口 main.py
from services.book_service import BookService
def main():
service = BookService()
while True:
print("\n" + "="*40)
print("图书管理系统")
print("="*40)
print("1. 显示所有图书")
print("2. 添加图书")
print("3. 修改图书")
print("4. 删除图书")
print("5. 搜索图书")
print("6. 查看图书详情")
print("0. 退出")
print("="*40)
choice = input("请选择操作:").strip()
if choice == "1":
books = service.get_all_books()
if books:
for book in books:
status = "可借" if book.available else "已借出"
print(f"\nID: {book.id}")
print(f"书名: {book.title}")
print(f"作者: {book.author}")
print(f"状态: {status}")
print("-"*30)
else:
print("暂无图书")
elif choice == "2":
try:
title = input("书名: ").strip()
author = input("作者: ").strip()
isbn = input("ISBN: ").strip()
year = int(input("出版年份: "))
price = float(input("价格: "))
book = service.create_book(title, author, isbn, year, price)
print(f"✅ 添加成功!图书ID: {book.id}")
except ValueError as e:
print(f"❌ {e}")
except Exception as e:
print(f"❌ 操作失败: {e}")
elif choice == "3":
try:
book_id = int(input("请输入要修改的图书ID: "))
print("(直接回车表示不修改)")
title = input("新书名: ").strip() or None
author = input("新作者: ").strip() or None
year_input = input("新出版年份: ").strip()
year = int(year_input) if year_input else None
price_input = input("新价格: ").strip()
price = float(price_input) if price_input else None
available_input = input("是否可借? (是/否): ").strip()
if available_input:
available = available_input.lower() == "是"
else:
available = None
book = service.update_book(book_id, title, author, year, price, available)
if book:
print(f"✅ 修改成功!")
else:
print("❌ 未找到该图书")
except ValueError as e:
print(f"❌ {e}")
elif choice == "4":
try:
book_id = int(input("请输入要删除的图书ID: "))
if service.delete_book(book_id):
print("✅ 删除成功!")
else:
print("❌ 未找到该图书")
except ValueError as e:
print(f"❌ {e}")
elif choice == "5":
keyword = input("请输入搜索关键词: ")
books = service.search_books(keyword)
if books:
for book in books:
status = "可借" if book.available else "已借出"
print(f"\nID: {book.id} | 书名: {book.title} | 作者: {book.author} | 状态: {status}")
else:
print("未找到相关图书")
elif choice == "6":
try:
book_id = int(input("请输入图书ID: "))
book = service.get_book(book_id)
if book:
status = "可借" if book.available else "已借出"
print(f"\n📚 图书详情")
print(f"ID: {book.id}")
print(f"书名: {book.title}")
print(f"作者: {book.author}")
print(f"ISBN: {book.isbn}")
print(f"出版年份: {book.year}")
print(f"价格: ¥{book.price:.2f}")
print(f"状态: {status}")
else:
print("❌ 未找到该图书")
except ValueError as e:
print(f"❌ {e}")
elif choice == "0":
print("👋 感谢使用,再见!")
break
else:
print("❌ 无效选择,请重试")
if __name__ == "__main__":
main()
运行说明
- 创建项目目录结构,将上述代码保存到对应文件中
- 运行程序:
python main.py
架构特点
- 分层清晰:数据模型、业务逻辑、数据访问分离
- 易于扩展:可以轻松添加新功能或更换存储方式
- 代码复用:各层职责明确,便于维护
- 分层依赖:上层依赖下层,避免循环引用
这个案例展示了Python中如何实现一个基本的分层架构,你可以根据实际需求进行扩展,比如添加数据库支持、Web API接口等。
标签: 依赖注入