semvec/pypi-packages huggingface dataset:
rpcx 0.4.0
Asynchronous RPC client/server with streaming support
License: MIT
Requires Python: <4.0,>=3.10

# RPCx

[![CI](https://github.com/uSpike/rpcx/actions/workflows/main.yml/badge.svg)](https://github.com/uSpike/rpcx/actions/workflows/main.yml)

Asynchronous RPC server/client for Python 3.10+ with streaming support.

- Async backend implemented by `anyio` providing support for `asyncio` and `trio`.
- Generic stream support for transport includes Websockets.
- Messages are serialized with `msgpack`.

```python
import math

import anyio
from anyio.streams.stapled import StapledObjectStream
from rpcx import RPCClient, RPCManager, RPCServer, Stream


async def add(a: int, b: int) -> int:
    """Add two numbers"""
    return a + b


async def fibonacci(n: int, stream: Stream) -> None:
    """Stream each number as computed in the Fibonacci sequence for a given starting number"""
    a, b = 0, 1

    for i in range(n):
        await stream.send(i)
        a, b = b, a + b


async def sum(stream: Stream) -> int:
    """Stream numbers from client and return the sum"""
    total = 0

    async for num in stream:
        total += num

    return total


manager = RPCManager()
manager.register("add", add)
manager.register("fibonacci", fibonacci)
manager.register("sum", sum)


async def main() -> None:
    # Create two connected stapled streams to simulate a network connection
    server_send, server_receive = anyio.create_memory_object_stream[bytes](math.inf)
    client_send, client_receive = anyio.create_memory_object_stream[bytes](math.inf)
    server_stream = StapledObjectStream(client_send, server_receive)
    client_stream = StapledObjectStream(server_send, client_receive)

    server = RPCServer(server_stream, manager)

    async with anyio.create_task_group() as task_group:
        task_group.start_soon(server.serve)

        async with RPCClient(client_stream) as client:
            # Simple method call
            assert await client.request("add", 1, 2) == 3

            # Streaming (server to client) example
            async with client.request_stream("fibonacci", 6) as stream:
                async for num in stream:
                    print(num)  # 1, 1, 2, 3, 5, 8

            # Streaming (client to server) example
            async with client.request_stream("sum") as stream:
                for num in range(10):
                    await stream.send(num)

            assert await stream == 45


anyio.run(main)
```

# Development

## Installation

Run `poetry install` to install the project.

## Execute tests

Execute `poetry run tox` to run tests for all python environments.

## Pre-commit hooks

Execute `poetry run pre-commit run -a` to run all linters, formatters, and checks.
factoria 0.1.0
A Python library to generate one or more Pydantic models with chainable state modifiers.
License: MIT
Requires Python: >=3.10

# Factoria

**Factoria** is a lightweight Python library for generating one or more [Pydantic](https://docs.pydantic.dev/) model instances with **chainable state modifiers**.  
It’s especially useful for testing, seeding, or creating predictable data scenarios.

## Features

- Generate single or multiple Pydantic models easily  
- Apply state transformations fluently
- Simple base class for building factories  
- Perfect for unit testing and data seeding  

## Installation

Using [Poetry](https://python-poetry.org/):

```bash
poetry add factoria
```

Or with pip

```bash
pip install factoria
```

## Examples

Create a Factory

```python
from pydantic import BaseModel
from factoria import BaseFactory


class User(BaseModel):
    id: int
    name: str
    email: str


class UserFactory(BaseFactory):
    def definition(self):
        return User(
            id=1,
            name="Alice",
            email="alice@example.com",
        )
```

Generate One or Many

```python
# Create one instance
user = UserFactory().make()
print(user)
# → User(id=1, name='Alice', email='alice@example.com')

# Create multiple instances
users = UserFactory().count(3).make()
print(len(users))
# → 3
```

You can define chainable states using the @apply decorator

```python
from datetime import datetime, timedelta
from pydantic import BaseModel
from factoria import BaseFactory, apply


class Order(BaseModel):
    id: int
    customer: str
    total: float
    status: str = "pending"
    paid_at: datetime | None = None
    shipped_at: datetime | None = None
    cancelled: bool = False


class OrderFactory(BaseFactory):

    def definition(self):
        return Order(
            id=1,
            customer="John Doe",
            total=199.90,
        )

    @apply
    def paid(self, obj):
        obj.status = "paid"
        obj.paid_at = datetime.utcnow()

    @apply
    def shipped(self, obj):
        obj.status = "shipped"
        obj.shipped_at = datetime.utcnow() + timedelta(days=1)

    @apply
    def cancelled(self, obj):
        obj.status = "cancelled"
        obj.cancelled = True
```

Some modifiers examples

```python
# One paid order
paid_order = OrderFactory().paid().make()

# Two shipped orders
shipped_orders = OrderFactory().count(2).shipped().make()

# Three cancelled orders
cancelled_orders = OrderFactory().count(3).cancelled().make()

# A complex example (paid and shipped)
paid_and_shipped = OrderFactory().count(2).paid().shipped().make()
```
holly-py 0.1.0
一个简单的 HelloWorld 示例 Python 包
License: MIT
Requires Python: >=3.8
Keywords: hello, world, demo, example

# Holly-Py

一个简单的 HelloWorld 示例 Python 包，用于演示如何创建、测试和发布 Python 包。

## 功能特性

- 简单的问候函数
- 面向对象的 HelloWorld 类
- 完整的单元测试
- 类型提示支持
- 文档字符串

## 安装

### 从 PyPI 安装（发布后）

```bash
pip install holly-py
```

### 从源码安装

```bash
git clone https://github.com/yourusername/holly-py.git
cd holly-py
pip install -e .
```

## 快速开始

### 使用函数

```python
from holly_py import greet

# 默认问候
print(greet())  # 输出: Hello, World!

# 自定义名字
print(greet("Python"))  # 输出: Hello, Python!
```

### 使用类

```python
from holly_py import HelloWorld

# 创建实例
hw = HelloWorld("开发者")

# 简单问候
print(hw.greet())  # 输出: Hello, 开发者!

# 多次问候
print(hw.greet_multiple(3))
# 输出:
# Hello, 开发者!
# Hello, 开发者!
# Hello, 开发者!
```

## 开发

### 安装开发依赖

```bash
pip install -e ".[dev]"
```

### 运行测试

```bash
# 运行所有测试
pytest

# 运行测试并显示覆盖率
pytest --cov=holly_py

# 运行特定测试文件
pytest tests/test_core.py -v
```

### 代码格式化

```bash
# 使用 black 格式化代码
black holly_py tests

# 检查代码风格
flake8 holly_py tests

# 类型检查
mypy holly_py
```

## 项目结构

```
holly-py/
├── holly_py/             # 主包目录
│   ├── __init__.py       # 包初始化文件
│   └── core.py           # 核心功能模块
├── tests/                # 测试目录
│   ├── __init__.py
│   └── test_core.py      # 核心功能测试
├── pyproject.toml        # 项目配置文件（推荐）
├── setup.py              # 安装脚本（向后兼容）
├── README.md             # 项目说明文档
├── LICENSE               # 许可证文件
└── .gitignore           # Git 忽略文件
```

## API 文档

### `greet(name: str = "World") -> str`

生成问候语。

**参数:**
- `name` (str): 要问候的名字，默认为 "World"

**返回:**
- str: 问候语字符串

### `HelloWorld`

HelloWorld 类提供面向对象的问候功能。

**方法:**
- `__init__(name: str = "World")`: 初始化实例
- `greet() -> str`: 返回问候语
- `greet_multiple(times: int = 1) -> str`: 返回多次问候语

## 许可证

MIT License - 详见 [LICENSE](LICENSE) 文件

## 贡献

欢迎提交 Issue 和 Pull Request！

## 作者

Your Name - your.email@example.com

## 更新日志

### 0.1.0 (2025-11-13)
- 初始版本
- 实现基本的 greet 函数
- 实现 HelloWorld 类
- 添加完整的单元测试