Zhang XiJun

#安装
pip install python-dotenv

#.env

db_host=localhost
db_port=3306
db_user=root
db_password=123456
db_name=test

from dotenv import load_dotenv
load_dotenv()

import os

db_host = os.getenv("db_host")
db_port = os.getenv("db_port")
db_user = os.getenv("db_user")
db_password = os.getenv("db_password")
db_name = os.getenv("db_name")

print(db_host)
print(db_port)
print(db_user)
print(db_password)
print(db_name)

# Environment variables
*.env

a = [1, 2, 3]      # ob_refcnt = 1
b = a              # ob_refcnt = 2
c = [a, "hello"]   # ob_refcnt = 3（因为 c[0] 也引用了它）
del b              # ob_refcnt = 2

操作系统
 └── 进程（Process） ← 资源容器（内存、文件描述符等）
      └── 线程（Thread） ← CPU 调度单元（至少一个）
           └── 协程（Coroutine） ← 用户态逻辑任务（可多个，协作式切换）

from time import sleep, perf_counter

def fetch_url(url):
    print('Fetching the URL')
    sleep(1)#模拟阻塞
    print('Finished fetching')
    return 'url_content'

def read_file(filepath):
    print('Reading the file')
    sleep(1)#模拟阻塞
    print('Finished reading')
    return 'file_content'

def main():
    url = 'example.com'
    filepath = 'example.txt'
    fetch_result = fetch_url(url)
    read_result = read_file(filepath)

if __name__ == '__main__':
    start_time = perf_counter()
    main()
    end_time = perf_counter()
    print(f'Time taken: {end_time - start_time:.2f} seconds')

Fetching the URL
Finished fetching
Reading the file
Finished reading
Time taken: 2.00 seconds

from time import sleep, perf_counter
import asyncio

#定义协程函数
async def fetch_url(url):
    print(f'Fetching the url')
    #把阻塞操作包装成协程
    await asyncio.sleep(1)
    print('Finished fetching')
    return 'url_content'

async def read_file(filepath):
    print('Reading the file')
    await asyncio.sleep(1)
    print('Finished reading')
    return 'file_content'

#若想使用await，需要把main函数定义成协程函数
async def main():
    url = 'example.com'
    filepath = 'example.txt'
    #创建任务
    tasks = [asyncio.create_task(coro) for coro in [
        fetch_url(url),
        read_file(filepath)]]
    fetch_result = await tasks[0]
    read_result = await tasks[1]

if __name__ == '__main__':
    start_time = perf_counter()
    #创建事件循环
    asyncio.run(main())
    end_time = perf_counter()
    print(f'Time taken: {end_time - start_time:.2f} seconds')

Fetching the url
Reading the file
Finished fetching
Finished reading
Time taken: 1.01 seconds

import asyncio

#协程函数
async def coroutine_func():
    return 'coroutine_result'

def main():
    print(coroutine_func())
    result = asyncio.run(coroutine_func())
    print("---")
    print(result)
    

if __name__ == '__main__':
    main()

<coroutine object coroutine_func at 0x000001958D88A4B0>
d:\code\python\learn_pythontips\learn_async\coroutine_func.py:8: RuntimeWarning: coroutine 'coroutine_func' was never awaited
  print(coroutine_func())
RuntimeWarning: Enable tracemalloc to get the object allocation traceback
---
coroutine_result

import asyncio

async def work(name, delay):
    print(f"开始 {name}")
    await asyncio.sleep(delay)
    print(f"完成 {name}")
    return f"result-{name}"

async def main():
    # 创建任务 → 立即开始执行！
    task1 = asyncio.create_task(work("A", 2))
    task2 = asyncio.create_task(work("B", 1))

    # 等待任务完成（可获取结果）
    r1 = await task1
    r2 = await task2
    print(r1, r2)

asyncio.run(main())

开始 A
开始 B
完成 B
完成 A
result-A result-B

async def main():
    url = 'example.com'
    filepath = 'example.txt'
    results = await asyncio.gather(
        fetch_url(url),
        read_file(filepath))
    print(results)

Fetching the url
Reading the file
Finished fetching
Finished reading
['url_content', 'file_content']
Time taken: 1.00 seconds

results = asyncio.as_completed([
    fetch_url(url),
    read_file(filepath)])
for result in results:
    print(await result)

import asyncio
import time

def blocking_io():
    time.sleep(1)
    return "Done!"

async def main():
    result = await asyncio.to_thread(blocking_io)
    print(result)

asyncio.run(main())

import aiohttp
import asyncio

async def fetch(url):
    async with aiohttp.ClientSession() as session:
        async with session.get(url) as response:
            return await response.text()

asyncio.run(fetch("https://example.com"))

import aiofiles
import asyncio

async def main():
    # 写文件
    async with aiofiles.open('data.txt', 'w') as f:
        await f.write('Hello, async files!')

    # 读文件
    async with aiofiles.open('data.txt', 'r') as f:
        content = await f.read()
        print(content)

asyncio.run(main())

class DatabaseConnection:
    def __init__(self, host, port, username, password):
        self.host = host
        self.port = port
        self.username = username
        self.password = password

    def connect(self):
        return f"Connecting to database at {self.host}:{self.port} with username '{self.username}'"


def client():
    main_db = DatabaseConnection('localhost', 3306, 'root', 'password123')
    analytics_db = DatabaseConnection('192.168.1.1', 5432, 'admin', 'securepass')
    cache_db = DatabaseConnection('10.0.0.1', 27017, 'cacheuser', 'cachepass')

    print(main_db.connect())
    print(analytics_db.connect())
    print(cache_db.connect())


client()

def connection_factory(db_type):
    db_configs = {
        'main': {
            'host': 'localhost',
            'port': 3306,
            'username': 'root',
            'password': 'password123'
        },
        'analytics': {
            'host': '192.168.1.1',
            'port': 5432,
            'username': 'admin',
            'password': 'securepass'
        },
        'cache': {
            'host': '10.0.0.1',
            'port': 27017,
            'username': 'cacheuser',
            'password': 'cachepass'
        }
    }
    
    return DatabaseConnection(**db_configs[db_type])


# 测试工厂模式
if __name__ == "__main__":
    # 使用工厂函数创建不同类型的数据库连接
    main_db = connection_factory('main')
    analytics_db = connection_factory('analytics')
    cache_db = connection_factory('cache')
    
    # 测试连接
    print(main_db.connect())
    print(analytics_db.connect())
    print(cache_db.connect())

#factory_pattern.py
def connection_factory(db_type):
    """
    工厂函数：根据数据库类型创建相应的数据库连接
    
    Args:
        db_type (str): 数据库类型 ('main', 'analytics', 'cache')
    
    Returns:
        DatabaseConnection: 数据库连接实例
    
    Raises:
        KeyError: 当提供的数据库类型不存在时
    """
    from config import DATABASE_CONFIGS
    if db_type not in DATABASE_CONFIGS:
        raise ValueError(f"Unknown database type: {db_type}. Available types: {list(DATABASE_CONFIGS.keys())}")
    
    return DatabaseConnection(**DATABASE_CONFIGS[db_type])

#config.py
# 数据库配置字典
DATABASE_CONFIGS = {
    'main': {
        'host': 'localhost',
        'port': 3306,
        'username': 'root',
        'password': 'password123'
    },
    'analytics': {
        'host': '192.168.1.1',
        'port': 5432,
        'username': 'admin',
        'password': 'securepass'
    },
    'cache': {
        'host': '10.0.0.1',
        'port': 27017,
        'username': 'cacheuser',
        'password': 'cachepass'
    }
}

class DatabaseConnectionBuilder:
    """数据库连接建造者类"""
    
    def __init__(self,host,port,username,password):
        self._config = {
            'host': host,
            'port': port,
            'username': username,
            'password': password,
        }
        
    def set_max_connections(self, max_connections):
        """设置最大连接数"""
        if max_connections is not None and max_connections <= 0:
            raise ValueError("Max connections must be positive")
        self._config['max_connections'] = max_connections
        return self
    
    def set_timeout(self, timeout):
        """设置超时时间"""
        if timeout is not None and timeout <= 0:
            raise ValueError("Connect timeout must be positive")
        self._config['timeout'] = timeout
        return self
        
    def build(self):
        """构建最终的数据库连接对象"""  
        return DatabaseConnection(**self._config)
        
        
# 演示建造者模式的使用
if __name__ == "__main__":
    # 创建一个数据库连接建造者
    builder = DatabaseConnectionBuilder(
        host='localhost',
        port=3306,
        username='root',
        password='password123'
    )
    
    # 使用建造者模式构建数据库连接
    connection = builder.set_max_connections(100) \
                        .set_timeout(30) \
                        .enable_ssl() \
                        .set_connection_pool('my_pool') \
                        .set_retry_attempts(3) \
                        .enable_compression() \
                        .set_read_preference('secondaryPreferred') \
                        .build()

class DatabaseConnection:
    """数据库连接单例类"""
    
    _instance = None
    _lock = threading.Lock()  # 线程锁，确保线程安全
    
    def __new__(cls, *args, **kwargs):
        """重写__new__方法实现单例模式"""
        if cls._instance is None:
            with cls._lock:  # 使用双重检查锁定
                if cls._instance is None:
                    cls._instance = super().__new__(cls)
        return cls._instance
    
    def __init__(self, host, port, username, password):
        """初始化数据库连接参数"""
        # 防止重复初始化
        if hasattr(self, '_initialized'):
            return
        
        self.host = host
        self.port = port
        self.username = username
        self.password = password
        self._initialized = True
    
    def connect(self):
        """连接数据库"""
        return f"Connecting to database at {self.host}:{self.port} with username '{self.username}'"

def client():
    db1 = DatabaseConnection(host='localhost', port=3306, username='root', password='password')
    db2 = DatabaseConnection(host='localhost', port=3306, username='root', password='password')
    print(db1 is db2)

client()

pip install package_name.whl

[build-system]
requires = ["hatchling"]
build-backend = "hatchling.build"

pip install .
# 或
python -m build

my_project/
├── pyproject.toml
├── src/
│   └── my_utils/               ← 你写的包
│       ├── __init__.py
│       └── math.py
├── scripts/
│   └── run_demo.py             ← 想在这里用 my_utils

# 👇 新增：显式声明包位置
[tool.hatch.build.targets.wheel]
packages = ["src/my_utils"]

uv pip install -e .

my_project/
├── my_package/
│   ├── __init__.py
│   └── module.py
├── tests/
├── setup.py
└── README.md

my_project/
├── src/
│   └── my_package/
│       ├── __init__.py
│       └── module.py
├── tests/
├── pyproject.toml
└── README.md

dist/my_package-0.1.0-py3-none-any.whl
dist/my_package-0.1.0.tar.gz

uv pip install -e .

import sys
print(sys.path)

─src
│  └─pytest_demo
│          calculator.py
│          __init__.py

"""
pytest_demo package
"""

from .calculator import Calculator

__all__ = ['Calculator']

└─tests
    │  test_calculator.py

from pytest_demo.calculator import Calculator

def test_add() -> None:
    calc = Calculator()
    assert calc.add(2, 1) == 3


def test_div() -> None:
    calc = Calculator()
    assert calc.divide(2, 1) == 2

import pytest

def test_divide_by_zero() -> None:
    calc = Calculator()
    with pytest.raises(ZeroDivisionError):
        calc.divide(2, 0)

#工厂函数
@pytest.fixture
def calc() -> Calculator:
    return Calculator()

def test_add(calc: Calculator) -> None:
    assert calc.add(2, 1) == 3

@pytest.fixture(scope="session")  # 整个测试会话只启动一次
def browser():
    driver = webdriver.Chrome()
    yield driver
    driver.quit()

def test_login(browser):
    browser.get("/login")
def test_profile(browser):
    browser.get("/profile")

pytest.fixture
def temp_file():
    # setup: 创建临时文件
    path = "temp.txt"
    with open(path, "w") as f:
        f.write("hello")
    
    yield path  # 测试函数在此处获得 path

    # teardown: 清理
    os.remove(path)

def test_read_file(temp_file):
    with open(temp_file) as f:
        assert f.read() == "hello"

@field_validator("name")
def _validate_name(cls, v: str) -> str:#cls 表示当前模型类
    v = str(v).strip()
    if not v:
        raise ValueError("name 不能为空")
    return v

pip install "lightrag-hku[api]"
cp env.example .env
lightrag-server

pip install lightrag-hku

LLM_BINDING=openai
LLM_MODEL=qwen3-max-preview
LLM_BINDING_HOST=https://dashscope.aliyuncs.com/compatible-mode/v1
LLM_BINDING_API_KEY=sk-

EMBEDDING_BINDING=openai
EMBEDDING_MODEL=text-embedding-v4
EMBEDDING_DIM=1024
EMBEDDING_BINDING_API_KEY=sk-
EMBEDDING_BINDING_HOST=https://dashscope.aliyuncs.com/compatible-mode/v1

############################
### 数据存储选择
############################
### 默认存储（推荐用于小规模部署）
LIGHTRAG_KV_STORAGE=JsonKVStorage
LIGHTRAG_DOC_STATUS_STORAGE=JsonDocStatusStorage
LIGHTRAG_GRAPH_STORAGE=NetworkXStorage
LIGHTRAG_VECTOR_STORAGE=NanoVectorDBStorage

### 图存储（推荐用于生产部署）
# LIGHTRAG_GRAPH_STORAGE=Neo4JStorage
# LIGHTRAG_GRAPH_STORAGE=MemgraphStorage

### PostgreSQL
# LIGHTRAG_KV_STORAGE=PGKVStorage
# LIGHTRAG_DOC_STATUS_STORAGE=PGDocStatusStorage
# LIGHTRAG_GRAPH_STORAGE=PGGraphStorage
# LIGHTRAG_VECTOR_STORAGE=PGVectorStorage

### PostgreSQL 配置
# POSTGRES_HOST=localhost
# POSTGRES_PORT=5432
# POSTGRES_USER=您的用户名
# POSTGRES_PASSWORD='您的密码'
# POSTGRES_DATABASE=您的数据库

### Neo4j 配置
# NEO4J_URI=neo4j+s://xxxxxxxx.databases.neo4j.io
# NEO4J_USERNAME=neo4j
# NEO4J_PASSWORD='您的密码'
# NEO4J_DATABASE=noe4j

### Milvus 配置
# MILVUS_URI=http://localhost:19530
# MILVUS_DB_NAME=lightrag
# MILVUS_USER=root
# MILVUS_PASSWORD=您的密码

lightrag-server

# 启动实例1
lightrag-server --port 9621 --workspace space1

# 启动实例2
lightrag-server --port 9622 --workspace space2

LIGHTRAG_KV_STORAGE=PGKVStorage
LIGHTRAG_VECTOR_STORAGE=PGVectorStorage
LIGHTRAG_GRAPH_STORAGE=PGGraphStorage
LIGHTRAG_DOC_STATUS_STORAGE=PGDocStatusStorage

class LightRAGStorage:
    """LightRAG存储和检索类

    支持多种存储后端：
    - KV存储: PostgreSQL
    - 文档状态存储: PostgreSQL
    - 图存储: Neo4j
    - 向量存储: Milvus

    使用workspace实现数据隔离
    """

    def __init__(self, workspace: str = "default"):
        """初始化LightRAG存储

        Args:
            workspace: 工作空间名称，用于数据隔离
        """
        self.workspace = workspace

        # 加载环境变量
        load_dotenv()

        # 使用文件同级的 lightrag_storage 目录
        self.working_dir = os.path.join(os.path.dirname(__file__), "lightrag_storage")

        self.rag: Optional[LightRAG] = None

        # 确保工作目录存在
        os.makedirs(self.working_dir, exist_ok=True)

    async def _get_llm_model_func(self):
        """LLM模型函数"""
        async def llm_model_func(
            prompt,
            system_prompt=None,
            history_messages=[],
            keyword_extraction=False,
            **kwargs
        ) -> str:
            return await openai_complete_if_cache(
                model=os.getenv("DASHSCOPE_MODEL", "qwen-max"),
                prompt=prompt,
                system_prompt=system_prompt,
                history_messages=history_messages,
                api_key=os.getenv("DASHSCOPE_API_KEY"),
                base_url=os.getenv("DASHSCOPE_API_BASE", "https://dashscope.aliyuncs.com/compatible-mode/v1"),
                **kwargs
            )
        return llm_model_func

    async def _get_embedding_func(self):
        """嵌入模型函数"""
        async def embedding_func(texts: List[str]) -> np.ndarray:
            return await openai_embed(
                texts,
                model=os.getenv("DASHSCOPE_EMBEDDING_MODEL", "text-embedding-v4"),
                api_key=os.getenv("DASHSCOPE_API_KEY"),
                base_url=os.getenv("DASHSCOPE_API_BASE", "https://dashscope.aliyuncs.com/compatible-mode/v1")
            )
        return embedding_func


    async def initialize(self) -> None:
        """初始化LightRAG实例"""
        if self.rag is not None:
            return

        # 获取模型函数
        llm_model_func = await self._get_llm_model_func()
        embedding_func = await self._get_embedding_func()

        # 存储配置 - 统一使用字符串方式，让LightRAG自动处理
        # 图存储配置
        graph_storage = os.getenv("LIGHTRAG_GRAPH_STORAGE", "Neo4JStorage")

        # KV存储配置
        kv_storage = os.getenv("LIGHTRAG_KV_STORAGE", "PGKVStorage")

        # 文档状态存储配置
        doc_status_storage = os.getenv("LIGHTRAG_DOC_STATUS_STORAGE", "PGDocStatusStorage")

        # 向量存储配置
        vector_storage = os.getenv("LIGHTRAG_VECTOR_STORAGE", "MilvusVectorDBStorage")

        # 创建LightRAG实例
        self.rag = LightRAG(
            working_dir=self.working_dir,
            embedding_func=EmbeddingFunc(
                func=embedding_func,
                embedding_dim=int(os.getenv("EMBEDDING_DIM", 1024))
            ),
            llm_model_func=llm_model_func,
            workspace=self.workspace,
            graph_storage=graph_storage,
            kv_storage=kv_storage,
            doc_status_storage=doc_status_storage,
            vector_storage=vector_storage
        )

        # 初始化存储
        await self.rag.initialize_storages()
        await initialize_pipeline_status()

    async def insert_text(self, text: str) -> None:
        """插入文本到LightRAG

        Args:
            text: 要插入的文本内容
        """
        if self.rag is None:
            await self.initialize()

        await self.rag.ainsert(text)

    async def insert_texts(self, texts: List[str]) -> None:
        """批量插入文本

        Args:
            texts: 文本列表
        """
        for text in texts:
            await self.insert_text(text)

    async def query(
        self,
        query: str,
        mode: str = "hybrid",
        **kwargs
    ) -> str:
        """执行查询

        Args:
            query: 查询文本
            mode: 查询模式 ("naive", "local", "global", "hybrid")
            **kwargs: 其他查询参数

        Returns:
            查询结果
        """
        if self.rag is None:
            await self.initialize()

        return await self.rag.aquery(
            query,
            param=QueryParam(mode=mode, **kwargs)
        )