Python高级编程核心技术：并发、内存管理与性能优化实战

Python作为一门功能强大且应用广泛的编程语言，在进阶阶段需要掌握更多核心技术来应对复杂场景。本文将从实际开发需求出发，重点讲解Python高级编程中的关键技术和实践方法。

对于已经掌握Python基础语法的开发者来说，进阶阶段需要关注的核心能力包括：并发编程优化、内存管理机制、元编程应用、性能调优策略、大型项目架构等。这些技术能够显著提升代码质量、执行效率和可维护性。

1. 核心能力速览

2. 适用场景与使用边界

Python高级技术主要适用于以下场景：

• 高并发网络服务开发（Web后端、API服务）
• 数据处理与科学计算（大数据分析、机器学习）
• 系统工具与自动化脚本（运维自动化、测试框架）
• 桌面应用与游戏开发（GUI程序、小游戏）

使用边界说明：

• CPU密集型任务建议使用多进程或C扩展
• 内存敏感场景需要谨慎管理对象生命周期
• 实时性要求极高的场景可能需结合其他语言
• 大型企业级项目需要严格的代码规范和架构设计

3. 环境准备与前置条件

3.1 Python版本选择

推荐使用Python 3.8及以上版本，新版本在性能和改进方面有显著提升：

# 检查当前Python版本 python --version # 或 python3 --version # 安装最新稳定版（以Ubuntu为例） sudo apt update sudo apt install python3.11 python3.11-venv

3.2 开发环境配置

使用虚拟环境隔离项目依赖：

# 创建虚拟环境 python3 -m venv advanced_python_env # 激活虚拟环境（Linux/macOS） source advanced_python_env/bin/activate # 激活虚拟环境（Windows） advanced_python_env\Scripts\activate # 安装基础工具包 pip install ipython black flake8 mypy pytest

3.3 必备工具栈

• 代码编辑器: VSCode + Python扩展 或 PyCharm Professional
• 版本控制: Git + GitHub/GitLab
• 包管理: pip + requirements.txt 或 Poetry
• 测试框架: pytest + coverage
• 文档工具: Sphinx 或 MkDocs

4. 并发编程深度解析

4.1 多线程实战应用

Python的多线程适合I/O密集型任务，但由于GIL的存在，不适合CPU密集型计算：

import threading import time from concurrent.futures import ThreadPoolExecutor def io_bound_task(task_id, duration): """模拟I/O密集型任务""" print(f"任务 {task_id} 开始执行") time.sleep(duration) # 模拟I/O等待 print(f"任务 {task_id} 完成") return f"任务{task_id}_结果" # 使用线程池管理并发 def run_thread_pool_example(): with ThreadPoolExecutor(max_workers=3) as executor: # 提交多个任务 futures = [ executor.submit(io_bound_task, i, 2) for i in range(5) ] # 获取结果 results = [future.result() for future in futures] print("所有任务完成:", results) if __name__ == "__main__": run_thread_pool_example()

4.2 多进程解决GIL限制

对于CPU密集型任务，使用多进程绕过GIL限制：

import multiprocessing import math from concurrent.futures import ProcessPoolExecutor def cpu_intensive_task(n): """CPU密集型任务：计算素数""" primes = [] for num in range(2, n + 1): if all(num % i != 0 for i in range(2, int(math.sqrt(num)) + 1)): primes.append(num) return primes def run_process_pool_example(): # 获取CPU核心数 cpu_count = multiprocessing.cpu_count() print(f"可用CPU核心数: {cpu_count}") with ProcessPoolExecutor(max_workers=cpu_count) as executor: # 分配不同规模的任务 tasks = [10000, 20000, 15000, 25000] futures = [executor.submit(cpu_intensive_task, n) for n in tasks] results = [future.result() for future in futures] print(f"计算完成，最大素数列表长度: {max(len(r) for r in results)}") if __name__ == "__main__": run_process_pool_example()

4.3 异步编程实战

asyncio库为高并发I/O操作提供了更好的解决方案：

import asyncio import aiohttp import time async def fetch_url(session, url): """异步获取URL内容""" try: async with session.get(url, timeout=10) as response: content = await response.text() return len(content) except Exception as e: print(f"请求 {url} 失败: {e}") return 0 async def run_async_demo(): urls = [ "https://httpbin.org/delay/1", "https://httpbin.org/delay/2", "https://httpbin.org/delay/1", "https://httpbin.org/delay/3" ] async with aiohttp.ClientSession() as session: tasks = [fetch_url(session, url) for url in urls] results = await asyncio.gather(*tasks) print(f"异步获取完成，结果: {results}") def benchmark_async_vs_sync(): """对比异步和同步的性能差异""" async def async_test(): start = time.time() await run_async_demo() return time.time() - start # 运行测试 elapsed = asyncio.run(async_test()) print(f"异步执行耗时: {elapsed:.2f}秒") if __name__ == "__main__": benchmark_async_vs_sync()

5. 内存管理机制深入

5.1 引用计数与垃圾回收

理解Python的内存管理机制有助于避免内存泄漏：

import sys import gc class MemoryDemo: def __init__(self, name): self.name = name print(f"创建对象: {self.name}") def __del__(self): print(f"销毁对象: {self.name}") def demonstrate_reference_counting(): """演示引用计数机制""" obj1 = MemoryDemo("obj1") print(f"obj1引用计数: {sys.getrefcount(obj1)}") obj2 = obj1 # 增加引用 print(f"obj1引用计数: {sys.getrefcount(obj1)}") del obj2 # 减少引用 print(f"obj1引用计数: {sys.getrefcount(obj1)}") del obj1 # 引用计数为0，触发销毁 def demonstrate_circular_reference(): """演示循环引用问题""" class Node: def __init__(self, name): self.name = name self.next = None # 创建循环引用 node1 = Node("节点1") node2 = Node("节点2") node1.next = node2 node2.next = node1 print("循环引用创建完成") print(f"垃圾回收阈值: {gc.get_threshold()}") # 强制垃圾回收 collected = gc.collect() print(f"回收的垃圾对象数: {collected}") if __name__ == "__main__": demonstrate_reference_counting() print("\n" + "="*50 + "\n") demonstrate_circular_reference()

5.2 内存优化技巧

import tracemalloc from memory_profiler import profile class OptimizedDataProcessor: """优化内存使用的数据处理类""" def __init__(self): self._cache = {} self._large_data = None def process_large_dataset(self, data): """处理大数据集的内存优化方法""" # 使用生成器避免一次性加载所有数据 def data_generator(): for item in data: yield self._process_item(item) # 分批处理 batch_size = 1000 batch = [] for result in data_generator(): batch.append(result) if len(batch) >= batch_size: yield from self._process_batch(batch) batch = [] if batch: yield from self._process_batch(batch) def _process_item(self, item): """处理单个数据项""" return item * 2 def _process_batch(self, batch): """处理批次数据""" return batch @profile def memory_usage_demo(): """内存使用分析示例""" processor = OptimizedDataProcessor() # 生成测试数据 large_data = list(range(100000)) # 处理数据 results = list(processor.process_large_dataset(large_data)) return len(results) if __name__ == "__main__": # 启动内存跟踪 tracemalloc.start() result = memory_usage_demo() print(f"处理完成，结果数量: {result}") # 显示内存使用情况 snapshot = tracemalloc.take_snapshot() top_stats = snapshot.statistics('lineno') print("\n内存使用排名前5:") for stat in top_stats[:5]: print(stat) tracemalloc.stop()

6. 元编程高级应用

6.1 装饰器深度应用

from functools import wraps import time import logging from typing import Any, Callable def retry(max_attempts: int = 3, delay: float = 1.0): """重试装饰器""" def decorator(func: Callable) -> Callable: @wraps(func) def wrapper(*args, **kwargs) -> Any: last_exception = None for attempt in range(max_attempts): try: return func(*args, **kwargs) except Exception as e: last_exception = e if attempt < max_attempts - 1: print(f"尝试 {attempt + 1} 失败，{delay}秒后重试: {e}") time.sleep(delay) else: print(f"所有尝试失败，最后错误: {e}") raise last_exception return wrapper return decorator def timing(func: Callable) -> Callable: """计时装饰器""" @wraps(func) def wrapper(*args, **kwargs) -> Any: start_time = time.time() try: return func(*args, **kwargs) finally: end_time = time.time() print(f"函数 {func.__name__} 执行时间: {end_time - start_time:.4f}秒") return wrapper class CacheDecorator: """缓存装饰器类""" def __init__(self, max_size: int = 100): self.cache = {} self.max_size = max_size self.hits = 0 self.misses = 0 def __call__(self, func: Callable) -> Callable: @wraps(func) def wrapper(*args, **kwargs) -> Any: # 生成缓存键 key = self._make_key(args, kwargs) if key in self.cache: self.hits += 1 return self.cache[key] self.misses += 1 result = func(*args, **kwargs) # 缓存管理 if len(self.cache) >= self.max_size: # 简单的LRU策略：移除第一个项目 first_key = next(iter(self.cache)) del self.cache[first_key] self.cache[key] = result return result return wrapper def _make_key(self, args, kwargs) -> tuple: """生成缓存键""" key = args + tuple(sorted(kwargs.items())) return key def cache_info(self) -> dict: """返回缓存统计信息""" return { 'hits': self.hits, 'misses': self.misses, 'size': len(self.cache), 'hit_rate': self.hits / (self.hits + self.misses) if (self.hits + self.misses) > 0 else 0 } # 使用装饰器组合 @timing @retry(max_attempts=3, delay=0.5) def expensive_operation(x: int, y: int) -> int: """模拟昂贵操作""" if x < 0: raise ValueError("x不能为负数") time.sleep(0.1) # 模拟耗时操作 return x * y # 创建缓存实例 cacher = CacheDecorator(max_size=50) @cacher def fibonacci(n: int) -> int: """计算斐波那契数列（使用缓存优化）""" if n <= 1: return n return fibonacci(n-1) + fibonacci(n-2) def demonstrate_decorators(): """演示装饰器使用""" print("=== 重试装饰器演示 ===") try: result = expensive_operation(5, 3) print(f"结果: {result}") except Exception as e: print(f"操作失败: {e}") print("\n=== 缓存装饰器演示 ===") start = time.time() fib_result = fibonacci(35) end = time.time() print(f"斐波那契(35) = {fib_result}, 耗时: {end - start:.4f}秒") print(f"缓存统计: {cacher.cache_info()}") if __name__ == "__main__": demonstrate_decorators()

6.2 元类高级应用

class SingletonMeta(type): """单例模式元类""" _instances = {} def __call__(cls, *args, **kwargs): if cls not in cls._instances: cls._instances[cls] = super().__call__(*args, **kwargs) return cls._instances[cls] class DatabaseConnection(metaclass=SingletonMeta): """数据库连接单例类""" def __init__(self, connection_string: str): self.connection_string = connection_string self._connected = False print(f"初始化数据库连接: {connection_string}") def connect(self): """模拟连接数据库""" if not self._connected: self._connected = True print("数据库连接已建立") def disconnect(self): """模拟断开连接""" if self._connected: self._connected = False print("数据库连接已断开") class ValidatedMeta(type): """验证字段的元类""" def __new__(cls, name, bases, attrs): # 收集需要验证的字段 validated_fields = {} for attr_name, attr_value in attrs.items(): if isinstance(attr_value, Field): validated_fields[attr_name] = attr_value # 添加验证逻辑 if validated_fields: original_init = attrs.get('__init__') def new_init(self, *args, **kwargs): if original_init: original_init(self, *args, **kwargs) # 验证字段 for field_name, field in validated_fields.items(): value = getattr(self, field_name, None) if not field.validate(value): raise ValueError(f"字段 {field_name} 验证失败: {value}") attrs['__init__'] = new_init return super().__new__(cls, name, bases, attrs) class Field: """字段验证基类""" def __init__(self, field_type, required=True): self.field_type = field_type self.required = required def validate(self, value): """验证字段值""" if value is None: return not self.required return isinstance(value, self.field_type) class StringField(Field): """字符串字段""" def __init__(self, max_length=None, **kwargs): super().__init__(str, **kwargs) self.max_length = max_length def validate(self, value): if not super().validate(value): return False if value is not None and self.max_length and len(value) > self.max_length: return False return True class IntField(Field): """整数字段""" def __init__(self, min_value=None, max_value=None, **kwargs): super().__init__(int, **kwargs) self.min_value = min_value self.max_value = max_value def validate(self, value): if not super().validate(value): return False if value is not None: if self.min_value is not None and value < self.min_value: return False if self.max_value is not None and value > self.max_value: return False return True class UserModel(metaclass=ValidatedMeta): """使用验证元类的用户模型""" name = StringField(max_length=50, required=True) age = IntField(min_value=0, max_value=150, required=True) email = StringField(required=False) def __init__(self, name, age, email=None): self.name = name self.age = age self.email = email def demonstrate_metaclasses(): """演示元类应用""" print("=== 单例模式演示 ===") db1 = DatabaseConnection("mysql://localhost:3306/mydb") db2 = DatabaseConnection("mysql://localhost:3306/mydb") print(f"db1 is db2: {db1 is db2}") db1.connect() db2.connect() print("\n=== 验证元类演示 ===") try: user1 = UserModel("张三", 25, "zhangsan@example.com") print("用户1创建成功") user2 = UserModel("李四", 200) # 年龄超出范围 print("用户2创建成功") except ValueError as e: print(f"用户创建失败: {e}") if __name__ == "__main__": demonstrate_metaclasses()

7. 性能优化实战策略

7.1 代码剖析与优化

import cProfile import pstats from io import StringIO import numpy as np def inefficient_function(n): """低效的函数实现""" result = [] for i in range(n): row = [] for j in range(n): row.append(i * j) result.append(row) return result def optimized_function(n): """优化后的函数实现""" # 使用NumPy向量化操作 i = np.arange(n).reshape(-1, 1) j = np.arange(n) return i * j def profile_comparison(): """性能对比分析""" n = 1000 print("=== 性能对比分析 ===") # 分析低效版本 print("\n1. 低效版本分析:") profiler = cProfile.Profile() profiler.enable() result1 = inefficient_function(n) profiler.disable() stream = StringIO() stats = pstats.Stats(profiler, stream=stream) stats.sort_stats('cumulative') stats.print_stats(10) print(stream.getvalue()) # 分析优化版本 print("\n2. 优化版本分析:") profiler = cProfile.Profile() profiler.enable() result2 = optimized_function(n) profiler.disable() stream = StringIO() stats = pstats.Stats(profiler, stream=stream) stats.sort_stats('cumulative') stats.print_stats(10) print(stream.getvalue()) # 验证结果一致性 print("\n3. 结果验证:") array1 = np.array(result1) array2 = result2 print(f"结果是否一致: {np.array_equal(array1, array2)}") ### 7.2 算法优化实例 ```python from typing import List, Dict, Any import time def find_duplicates_naive(items: List[Any]) -> List[Any]: """查找重复项的朴素实现 O(n²)""" duplicates = [] n = len(items) for i in range(n): for j in range(i + 1, n): if items[i] == items[j] and items[i] not in duplicates: duplicates.append(items[i]) return duplicates def find_duplicates_optimized(items: List[Any]) -> List[Any]: """优化后的重复项查找 O(n)""" seen = set() duplicates = set() for item in items: if item in seen: duplicates.add(item) else: seen.add(item) return list(duplicates) def benchmark_duplicate_finding(): """性能基准测试""" # 生成测试数据 test_data = list(range(10000)) + [42, 123, 999] * 100 # 添加重复项 print("=== 重复项查找性能对比 ===") # 测试朴素实现 start_time = time.time() result1 = find_duplicates_naive(test_data) naive_time = time.time() - start_time # 测试优化实现 start_time = time.time() result2 = find_duplicates_optimized(test_data) optimized_time = time.time() - start_time print(f"朴素实现耗时: {naive_time:.4f}秒") print(f"优化实现耗时: {optimized_time:.4f}秒") print(f"性能提升: {naive_time/optimized_time:.1f}倍") print(f"结果一致性: {set(result1) == set(result2)}") ### 7.3 内存视图与缓冲区协议 ```python import array import memoryview def demonstrate_memory_views(): """演示内存视图的高效使用""" # 创建大型数组 large_array = array.array('i', range(1000000)) print("=== 内存视图性能演示 ===") # 传统切片操作（复制数据） start_time = time.time() slice_copy = large_array[100000:200000] slice_time = time.time() - start_time # 内存视图操作（零拷贝） start_time = time.time() mem_view = memoryview(large_array) slice_view = mem_view[100000:200000] view_time = time.time() - start_time print(f"传统切片耗时: {slice_time:.6f}秒") print(f"内存视图耗时: {view_time:.6f}秒") print(f"内存视图优势: {slice_time/view_time:.1f}倍") # 验证数据一致性 view_array = array.array('i', slice_view) print(f"数据一致性: {slice_copy == view_array}") if __name__ == "__main__": profile_comparison() print("\n" + "="*60 + "\n") benchmark_duplicate_finding() print("\n" + "="*60 + "\n") demonstrate_memory_views()

8. 大型项目架构实践

8.1 项目结构设计

advanced_python_project/ ├── src/ │ ├── package/ │ │ ├── __init__.py │ │ ├── core/ │ │ │ ├── __init__.py │ │ │ ├── processors.py │ │ │ └── validators.py │ │ ├── utils/ │ │ │ ├── __init__.py │ │ │ ├── loggers.py │ │ │ └── helpers.py │ │ └── api/ │ │ ├── __init__.py │ │ ├── routes.py │ │ └── middleware.py │ └── scripts/ │ ├── __init__.py │ ├── setup.py │ └── deploy.py ├── tests/ │ ├── __init__.py │ ├── unit/ │ │ ├── test_core.py │ │ └── test_utils.py │ └── integration/ │ └── test_api.py ├── docs/ │ ├── conf.py │ └── index.rst ├── requirements/ │ ├── base.txt │ ├── dev.txt │ └── prod.txt ├── .github/ │ └── workflows/ │ └── ci.yml └── configuration/ ├── logging.conf └── settings.py

8.2 配置管理最佳实践

# configuration/settings.py import os from typing import Dict, Any from dataclasses import dataclass @dataclass class DatabaseConfig: """数据库配置类""" host: str = "localhost" port: int = 5432 username: str = "user" password: str = "password" database: str = "app_db" @property def connection_string(self) -> str: return f"postgresql://{self.username}:{self.password}@{self.host}:{self.port}/{self.database}" @dataclass class APIConfig: """API配置类""" host: str = "0.0.0.0" port: int = 8000 debug: bool = False workers: int = 4 class Config: """主配置类""" def __init__(self): self.environment = os.getenv("ENVIRONMENT", "development") self.database = DatabaseConfig() self.api = APIConfig() self._load_environment_variables() def _load_environment_variables(self): """从环境变量加载配置""" # 数据库配置 if db_host := os.getenv("DB_HOST"): self.database.host = db_host if db_port := os.getenv("DB_PORT"): self.database.port = int(db_port) # API配置 if api_port := os.getenv("API_PORT"): self.api.port = int(api_port) if debug := os.getenv("DEBUG"): self.api.debug = debug.lower() == "true" # 全局配置实例 config = Config() def get_config() -> Config: """获取配置实例（支持依赖注入）""" return config

8.3 依赖注入容器

# src/core/dependency_injection.py from typing import Type, TypeVar, Dict, Any import inspect T = TypeVar('T') class DIContainer: """简单的依赖注入容器""" def __init__(self): self._singletons: Dict[Type, Any] = {} self._factories: Dict[Type, callable] = {} def register_singleton(self, interface: Type, implementation: Any): """注册单例""" self._singletons[interface] = implementation def register_factory(self, interface: Type, factory: callable): """注册工厂函数""" self._factories[interface] = factory def resolve(self, interface: Type[T]) -> T: """解析依赖""" # 检查单例 if interface in self._singletons: return self._singletons[interface] # 检查工厂 if interface in self._factories: return self._factories[interface]() # 尝试自动创建实例 return self._auto_create(interface) def _auto_create(self, interface: Type) -> Any: """自动创建实例""" if inspect.isabstract(interface): raise ValueError(f"无法创建抽象类 {interface} 的实例") # 获取构造函数的参数 signature = inspect.signature(interface.__init__) parameters = {} for name, param in signature.parameters.items(): if name == 'self': continue if param.annotation != inspect.Parameter.empty: parameters[name] = self.resolve(param.annotation) else: raise ValueError(f"参数 {name} 缺少类型注解") return interface(**parameters) # 使用示例 class DatabaseService: def __init__(self, config: DatabaseConfig): self.config = config def connect(self): print(f"连接到数据库: {self.config.connection_string}") class APIService: def __init__(self, config: APIConfig, db_service: DatabaseService): self.config = config self.db_service = db_service def start(self): self.db_service.connect() print(f"启动API服务: {self.config.host}:{self.config.port}") def demonstrate_di_container(): """演示依赖注入容器""" container = DIContainer() # 注册依赖 from configuration.settings import config container.register_singleton(DatabaseConfig, config.database) container.register_singleton(APIConfig, config.api) container.register_factory(DatabaseService, lambda: DatabaseService(container.resolve(DatabaseConfig))) # 解析服务 api_service = container.resolve(APIService) api_service.start() if __name__ == "__main__": demonstrate_di_container()

9. 测试策略与质量保证

9.1 单元测试最佳实践

# tests/unit/test_core.py import pytest from unittest.mock import Mock, patch from src.core.processors import DataProcessor class TestDataProcessor: """数据处理器测试类""" @pytest.fixture def processor(self): """创建处理器实例""" return DataProcessor() def test_process_data_valid_input(self, processor): """测试有效数据处理""" input_data = [1, 2, 3, 4, 5] expected = [2, 4, 6, 8, 10] result = processor.process_data(input_data, multiplier=2) assert result == expected def test_process_data_empty_input(self, processor): """测试空输入处理""" result = processor.process_data([]) assert result == [] @pytest.mark.parametrize("input_data,expected", [ ([1, 2, 3], [2, 4, 6]), ([], []), ([10], [20]), ]) def test_process_data_parametrized(self, processor, input_data, expected): """参数化测试""" result = processor.process_data(input_data, multiplier=2) assert result == expected @patch('src.core.processors.logging') def test_process_data_with_logging(self, mock_logging, processor): """测试日志记录""" input_data = [1, 2, 3] processor.process_data(input_data) # 验证日志调用 mock_logging.info.assert_called_once()

9.2 集成测试示例

# tests/integration/test_api.py import pytest from fastapi.testclient import TestClient from src.api.main import app class TestAPIIntegration: """API集成测试""" @pytest.fixture def client(self): """创建测试客户端""" return TestClient(app) def test_health_endpoint(self, client): """测试健康检查端点""" response = client.get("/health") assert response.status_code == 200 assert response.json() == {"status": "healthy"} def test_data_processing_endpoint(self, client): """测试数据处理端点""" test_data = {"numbers": [1, 2, 3], "multiplier": 2} response = client.post("/process", json=test_data) assert response.status_code == 200 assert response.json() == {"result": [2, 4, 6]} def test_invalid_input_handling(self, client): """测试无效输入处理""" invalid_data = {"numbers": "not_a_list"} response = client.post("/process", json=invalid_data) assert response.status_code == 422 # 验证错误

9.3 性能测试策略

# tests/performance/test_performance.py import pytest import time from src.core.processors import DataProcessor @pytest.mark.performance class TestPerformance: """性能测试类""" @pytest.mark.benchmark def test_processing_performance(self): """数据处理性能测试""" processor = DataProcessor() large_dataset = list(range(100000)) start_time = time.time() result = processor.process_data(large_dataset) end_time = time.time() processing_time = end_time - start_time print(f"处理10万条数据耗时: {processing_time:.2f}秒") # 性能断言（根据实际需求调整阈值） assert processing_time < 5.0, "处理时间超出预期" # 验证结果正确性 assert len(result) == len(large_dataset) assert result[0] == 0 assert result[-1] == large_dataset[-1] @pytest.fixture(scope="session") def performance_threshold(): """性能阈值配置""" return { "max_processing_time": 5.0, "max_memory_usage": 100 # MB }

10. 部署与监控

10.1 Docker容器化部署

# Dockerfile FROM python:3.11-slim # 设置工作目录 WORKDIR /app # 复制依赖文件 COPY requirements/prod.txt . # 安装依赖 RUN pip install --no-cache-dir -r prod.txt # 复制应用代码 COPY src/ . # 创建非root用户 RUN useradd --create-home --shell /bin/bash app USER app # 暴露端口 EXPOSE 8000 # 启动命令 CMD ["python", "-m", "package.api.main"]

10.2 监控与日志配置

# src/utils/loggers.py import logging import logging.config import json from pathlib import Path def setup_logging(config_path: Path = None): """设置日志配置""" default_config = { 'version': 1, 'disable_existing_loggers': False, 'formatters': { 'standard': { 'format': '%(asctime)s - %(name)s - %(levelname)s - %(message)s' }, 'detailed': { 'format': '%(asctime)s - %(name)s - %(levelname)s - %(pathname)s:%(lineno)d - %(message)s' } }, 'handlers': { 'console': { 'class': 'logging.StreamHandler', 'formatter': 'standard',