Python中级教程 第七课:网络编程、并发与项目实践 6/7

浏览量:11 次 发布时间:2026-01-15 18:06 作者:明扬工控商城 下载docx

最近更新:Python中级教程 第三课:面向对象编程深入与装饰器


第六部分:项目结构与代码组织

6.1 项目结构示例

text

my_project/

├── README.md                 # 项目说明

├── LICENSE                   # 许可证

├── requirements.txt          # 依赖包列表

├── setup.py                  # 安装脚本

├── pyproject.toml           # 构建配置

├── .gitignore               # Git忽略文件

├── .env.example             # 环境变量示例

├── tests/                   # 测试目录

│   ├── __init__.py

│   ├── test_core.py

│   ├── test_api.py

│   └── test_integration.py

├── docs/                    # 文档目录

│   ├── index.md

│   ├── api.md

│   └── examples.md

├── examples/                # 示例目录

│   ├── basic_usage.py

│   └── advanced_usage.py

├── scripts/                 # 脚本目录

│   ├── setup_database.py

│   └── backup_data.py

├── src/                     # 源代码目录

│   └── my_package/         # 主包

│       ├── __init__.py

│       ├── core.py         # 核心功能

│       ├── models.py       # 数据模型

│       ├── api.py          # API接口

│       ├── utils.py        # 工具函数

│       ├── config.py       # 配置管理

│       └── cli.py          # 命令行接口

└── data/                   # 数据目录(可选)

   ├── input/

   └── output/

6.2 模块组织示例

python

# my_package/__init__.py

"""

my_package - 一个示例Python包

"""


__version__ = "1.0.0"

__author__ = "Your Name"

__email__ = "your.email@example.com"


# 导入关键功能,方便用户使用

from .core import main_function, helper_function

from .models import User, Product

from .api import APIClient

from .utils import setup_logging


# 初始化代码

def initialize():

   """初始化包"""

   import logging

   logging.getLogger(__name__).addHandler(logging.NullHandler())


# 自动初始化

initialize()


# my_package/config.py

import os

import json

from typing import Any, Dict, Optional

from pathlib import Path

import logging


class Config:

   """配置管理类"""

   

   # 默认配置

   DEFAULTS = {

       'debug': False,

       'log_level': 'INFO',

       'database': {

           'host': 'localhost',

           'port': 5432,

           'name': 'myapp',

           'user': 'postgres',

           'password': ''

       },

       'api': {

           'base_url': 'https://api.example.com',

           'timeout': 30,

           'retries': 3

       }

   }

   

   def __init__(self, config_file: Optional[str] = None):

       self.config_file = config_file

       self.config = self.DEFAULTS.copy()

       self.logger = logging.getLogger(__name__)

       

       # 加载配置文件

       if config_file:

           self.load_config(config_file)

       

       # 加载环境变量

       self.load_env_vars()

   

   def load_config(self, config_file: str):

       """从文件加载配置"""

       try:

           path = Path(config_file)

           if path.exists():

               with open(path, 'r') as f:

                   file_config = json.load(f)

                   self._deep_update(self.config, file_config)

               self.logger.info(f"已加载配置文件: {config_file}")

       except Exception as e:

           self.logger.warning(f"加载配置文件失败: {e}")

   

   def load_env_vars(self):

       """从环境变量加载配置"""

       # 数据库配置

       db_host = os.getenv('DB_HOST')

       if db_host:

           self.config['database']['host'] = db_host

       

       # API配置

       api_url = os.getenv('API_BASE_URL')

       if api_url:

           self.config['api']['base_url'] = api_url

       

       # 调试模式

       debug_env = os.getenv('DEBUG')

       if debug_env:

           self.config['debug'] = debug_env.lower() in ('true', '1', 'yes')

   

   def _deep_update(self, target: Dict, source: Dict):

       """深度更新字典"""

       for key, value in source.items():

           if key in target and isinstance(target[key], dict) and isinstance(value, dict):

               self._deep_update(target[key], value)

           else:

               target[key] = value

   

   def get(self, key: str, default: Any = None) -> Any:

       """获取配置值"""

       keys = key.split('.')

       value = self.config

       

       try:

           for k in keys:

               value = value[k]

           return value

       except (KeyError, TypeError):

           return default

   

   def set(self, key: str, value: Any):

       """设置配置值"""

       keys = key.split('.')

       config = self.config

       

       for k in keys[:-1]:

           if k not in config:

               config[k] = {}

           config = config[k]

       

       config[keys[-1]] = value

   

   def save(self, config_file: Optional[str] = None):

       """保存配置到文件"""

       save_file = config_file or self.config_file

       if not save_file:

           raise ValueError("未指定配置文件路径")

       

       try:

           path = Path(save_file)

           path.parent.mkdir(parents=True, exist_ok=True)

           

           with open(path, 'w') as f:

               json.dump(self.config, f, indent=2)

           

           self.logger.info(f"配置已保存到: {save_file}")

       except Exception as e:

           self.logger.error(f"保存配置失败: {e}")

           raise

   

   def __getitem__(self, key: str) -> Any:

       return self.get(key)

   

   def __setitem__(self, key: str, value: Any):

       self.set(key, value)


# 全局配置实例

_config = None


def get_config() -> Config:

   """获取全局配置实例"""

   global _config

   if _config is None:

       _config = Config()

   return _config


# my_package/utils.py

import logging

import sys

from datetime import datetime

from typing import Any, Callable, Optional

import functools

from pathlib import Path


def setup_logging(

   level: str = 'INFO',

   log_file: Optional[str] = None,

   format_string: Optional[str] = None

) -> logging.Logger:

   """设置日志配置"""

   

   if format_string is None:

       format_string = '%(asctime)s - %(name)s - %(levelname)s - %(message)s'

   

   formatter = logging.Formatter(format_string)

   

   # 根日志记录器

   root_logger = logging.getLogger()

   root_logger.setLevel(getattr(logging, level.upper()))

   

   # 清除现有处理器

   for handler in root_logger.handlers[:]:

       root_logger.removeHandler(handler)

   

   # 控制台处理器

   console_handler = logging.StreamHandler(sys.stdout)

   console_handler.setFormatter(formatter)

   root_logger.addHandler(console_handler)

   

   # 文件处理器

   if log_file:

       file_handler = logging.FileHandler(log_file, encoding='utf-8')

       file_handler.setFormatter(formatter)

       root_logger.addHandler(file_handler)

   

   return root_logger


def timer(func: Callable) -> Callable:

   """计时装饰器"""

   @functools.wraps(func)

   def wrapper(*args, **kwargs):

       start_time = datetime.now()

       result = func(*args, **kwargs)

       end_time = datetime.now()

       duration = end_time - start_time

       

       logger = logging.getLogger(func.__module__)

       logger.debug(f"函数 {func.__name__} 执行时间: {duration}")

       

       return result

   return wrapper


def retry(max_retries: int = 3, delay: float = 1.0, exceptions: tuple = (Exception,)):

   """重试装饰器"""

   def decorator(func: Callable) -> Callable:

       @functools.wraps(func)

       def wrapper(*args, **kwargs):

           last_exception = None

           

           for attempt in range(1, max_retries + 1):

               try:

                   return func(*args, **kwargs)

               except exceptions as e:

                   last_exception = e

                   

                   if attempt < max_retries:

                       import time

                       logger = logging.getLogger(func.__module__)

                       logger.warning(f"尝试 {attempt}/{max_retries} 失败: {e}, {delay}秒后重试...")

                       time.sleep(delay)

           

           raise last_exception

       return wrapper

   return decorator


def ensure_directory(path: str) -> Path:

   """确保目录存在"""

   path_obj = Path(path)

   path_obj.mkdir(parents=True, exist_ok=True)

   return path_obj


def format_size(size_bytes: int) -> str:

   """格式化文件大小"""

   for unit in ['B', 'KB', 'MB', 'GB', 'TB']:

       if size_bytes < 1024.0:

           return f"{size_bytes:.2f} {unit}"

       size_bytes /= 1024.0

   return f"{size_bytes:.2f} PB"


def safe_json_load(filepath: str, default: Any = None) -> Any:

   """安全加载JSON文件"""

   import json

   path = Path(filepath)

   

   if not path.exists():

       return default

   

   try:

       with open(path, 'r', encoding='utf-8') as f:

           return json.load(f)

   except (json.JSONDecodeError, UnicodeDecodeError) as e:

       logger = logging.getLogger(__name__)

       logger.error(f"加载JSON文件失败 {filepath}: {e}")

       return default


# my_package/cli.py

import argparse

import sys

import logging

from typing import List, Optional


class CLI:

   """命令行接口"""

   

   def __init__(self):

       self.parser = argparse.ArgumentParser(

           description="我的Python包命令行工具",

           formatter_class=argparse.RawDescriptionHelpFormatter,

           epilog="""

示例:

 %(prog)s run --input data.txt --output result.json

 %(prog)s config --set api.base_url https://api.example.com

           """

       )

       

       # 全局参数

       self.parser.add_argument(

           '--verbose', '-v',

           action='count',

           default=0,

           help='详细输出级别 (-v, -vv, -vvv)'

       )

       self.parser.add_argument(

           '--config',

           default='config.json',

           help='配置文件路径'

       )

       

       # 子命令

       subparsers = self.parser.add_subparsers(

           dest='command',

           title='命令',

           description='可用命令',

           required=True

       )

       

       # run命令

       run_parser = subparsers.add_parser('run', help='运行主程序')

       run_parser.add_argument('--input', '-i', required=True, help='输入文件')

       run_parser.add_argument('--output', '-o', help='输出文件')

       run_parser.add_argument('--workers', '-w', type=int, default=4, help='工作线程数')

       

       # config命令

       config_parser = subparsers.add_parser('config', help='管理配置')

       config_group = config_parser.add_mutually_exclusive_group(required=True)

       config_group.add_argument('--list', '-l', action='store_true', help='列出所有配置')

       config_group.add_argument('--get', '-g', help='获取配置值')

       config_group.add_argument('--set', '-s', nargs=2, metavar=('KEY', 'VALUE'), help='设置配置值')

       

       # version命令

       version_parser = subparsers.add_parser('version', help='显示版本信息')

       

       # setup命令

       setup_parser = subparsers.add_parser('setup', help='初始设置')

       setup_parser.add_argument('--force', '-f', action='store_true', help='强制重新设置')

   

   def setup_logging(self, verbosity: int):

       """根据详细级别设置日志"""

       level_map = {

           0: logging.WARNING,

           1: logging.INFO,

           2: logging.DEBUG,

           3: logging.DEBUG

       }

       

       level = level_map.get(min(verbosity, 3), logging.DEBUG)

       logging.basicConfig(level=level, format='%(levelname)s: %(message)s')

   

   def run_command(self, args):

       """运行命令"""

       from .core import main_function

       from .utils import ensure_directory

       

       logging.info(f"运行命令,输入: {args.input}")

       

       # 确保输出目录存在

       if args.output:

           output_path = ensure_directory(args.output).parent

       

       # 调用核心功能

       result = main_function(args.input, args.output, workers=args.workers)

       logging.info(f"运行完成,结果: {result}")

   

   def config_command(self, args):

       """配置命令"""

       from .config import get_config

       

       config = get_config()

       

       if args.list:

           import json

           print(json.dumps(config.config, indent=2))

       

       elif args.get:

           value = config.get(args.get)

           print(f"{args.get} = {value}")

       

       elif args.set:

           key, value = args.set

           # 尝试转换值类型

           try:

               # 如果是数字

               if '.' in value:

                   value = float(value)

               else:

                   value = int(value)

           except ValueError:

               # 保持为字符串

               pass

           

           config.set(key, value)

           config.save()

           print(f"已设置 {key} = {value}")

   

   def version_command(self, args):

       """版本命令"""

       from . import __version__

       print(f"my_package 版本 {__version__}")

   

   def setup_command(self, args):

       """设置命令"""

       logging.info("执行初始设置...")

       

       # 创建必要目录

       from .utils import ensure_directory

       

       directories = ['data', 'logs', 'cache']

       for directory in directories:

           ensure_directory(directory)

           logging.info(f"创建目录: {directory}")

       

       # 创建默认配置文件

       from .config import Config

       

       config = Config()

       config.save('config.json')

       

       logging.info("初始设置完成")

   

   def parse_and_run(self, argv: Optional[List[str]] = None):

       """解析参数并运行"""

       args = self.parser.parse_args(argv)

       

       # 设置日志

       self.setup_logging(args.verbose)

       

       # 根据命令执行相应函数

       if args.command == 'run':

           self.run_command(args)

       elif args.command == 'config':

           self.config_command(args)

       elif args.command == 'version':

           self.version_command(args)

       elif args.command == 'setup':

           self.setup_command(args)

       else:

           self.parser.print_help()

           return 1

       

       return 0


def main():

   """主函数"""

   cli = CLI()

   sys.exit(cli.parse_and_run())


if __name__ == '__main__':

   main()


# setup.py

from setuptools import setup, find_packages

import os


# 读取README

with open('README.md', 'r', encoding='utf-8') as f:

   long_description = f.read()


# 读取requirements.txt

with open('requirements.txt', 'r', encoding='utf-8') as f:

   requirements = [line.strip() for line in f if line.strip() and not line.startswith('#')]


setup(

   name="my-package",

   version="1.0.0",

   author="Your Name",

   author_email="your.email@example.com",

   description="一个示例Python包",

   long_description=long_description,

   long_description_content_type="text/markdown",

   url="https://github.com/yourusername/my-package",

   packages=find_packages(where="src"),

   package_dir={"": "src"},

   classifiers=[

       "Development Status :: 4 - Beta",

       "Intended Audience :: Developers",

       "License :: OSI Approved :: MIT License",

       "Operating System :: OS Independent",

       "Programming Language :: Python :: 3",

       "Programming Language :: Python :: 3.7",

       "Programming Language :: Python :: 3.8",

       "Programming Language :: Python :: 3.9",

       "Programming Language :: Python :: 3.10",

   ],

   python_requires=">=3.7",

   install_requires=requirements,

   extras_require={

       "dev": [

           "pytest>=6.0",

           "pytest-cov>=2.0",

           "black>=21.0",

           "flake8>=4.0",

           "mypy>=0.9",

       ],

       "docs": [

           "sphinx>=4.0",

           "sphinx-rtd-theme>=1.0",

       ],

   },

   entry_points={

       "console_scripts": [

           "my-package=my_package.cli:main",

       ],

   },

   include_package_data=True,

)


# requirements.txt

# 核心依赖

requests>=2.25.0

numpy>=1.20.0

pandas>=1.3.0

sqlalchemy>=1.4.0


# 可选依赖

# redis>=3.5.0

# pymongo>=3.11.0


# 测试依赖

# pytest>=6.0.0

# pytest-cov>=2.0.0


# 开发依赖

# black>=21.0

# flake8>=4.0


# pyproject.toml

[build-system]

requires = ["setuptools>=42", "wheel"]

build-backend = "setuptools.build_meta"


[project]

name = "my-package"

version = "1.0.0"

description = "一个示例Python包"

readme = "README.md"

requires-python = ">=3.7"

license = {text = "MIT License"}

authors = [

   {name = "Your Name", email = "your.email@example.com"}

]

classifiers = [

   "Development Status :: 4 - Beta",

   "Intended Audience :: Developers",

   "License :: OSI Approved :: MIT License",

   "Operating System :: OS Independent",

   "Programming Language :: Python :: 3",

]


[tool.black]

line-length = 88

target-version = ['py37', 'py38', 'py39', 'py310']


[tool.isort]

profile = "black"

multi_line_output = 3

include_trailing_comma = true


[tool.mypy]

python_version = "3.7"

warn_return_any = true

warn_unused_configs = true


[tool.pytest.ini_options]

testpaths = ["tests"]

python_files = ["test_*.py"]

python_classes = ["Test*"]

python_functions = ["test_*"]

addopts = "--verbose --color=yes"

6.3 代码质量工具




明扬工控商城

推荐阅读:

Python中级教程 第六课:高级特性与性能优化 2/2

Python中级教程 第六课:高级特性与性能优化 1/2

Python中级教程 第五课:异步编程与并发 2/2

Python中级教程 第五课:异步编程与并发 1/2

Python中级教程 第四课:上下文管理器、生成器与迭代器

Python中级教程 第三课:面向对象编程深入与装饰器

热门标签:
Python中级教程 第七课:网络编程、并发与项目实践 6/7.docx

将本文的Word文档下载到电脑

推荐度:

下载

全部评论

请登录
产业新闻-明扬资讯网
科技资讯-明扬资讯网