Yuanfen Python Library
Installation
pip install yuanfen
Config
Supports .json, .yaml / .yml, and .ini files. Instances are singletons per file path. Auto-reloads when the config file changes (uses polling by default).
from yuanfen import Config
config = Config("configs/config.yaml")
# Access via key (raises KeyError if missing)
print(config["app"]["name"])
# Access with default fallback
print(config.get("debug", False))
| Parameter | Type | Default | Description |
|---|---|---|---|
path |
str |
— | Path to the config file |
poll |
bool |
True |
Use polling observer (more compatible); set False for native FS events |
logger |
Logger |
Logger() |
Logger instance for reload messages |
Logger
Writes to both stdout and a daily-rotated file at logs/log (kept for 365 days).
import logging
from yuanfen import Logger
logger = Logger(name="my-service", level=logging.DEBUG)
logger.debug("debug log")
logger.info("info log")
logger.warning("warning log")
logger.error("error log")
| Parameter | Type | Default | Description |
|---|---|---|---|
name |
str |
None |
Prefix added to every log message as [name] |
level |
int |
logging.INFO |
Minimum log level |
logger |
logging.Logger |
root logger | Use a custom logger instance |
Response
Pydantic response models for FastAPI.
from yuanfen import BaseResponse, SuccessResponse, ErrorResponse
# Default: code=0, message="SUCCESS"
SuccessResponse(data={"id": 1})
# Default: code=1000, message="ERROR"
ErrorResponse(message="Something went wrong")
# Custom
BaseResponse(code=404, message="Not found")
import uvicorn
from fastapi import FastAPI
from yuanfen import SuccessResponse, ErrorResponse
app = FastAPI()
@app.get("/health-check")
def health_check():
return SuccessResponse(data="OK")
@app.get("/items/{item_id}")
def get_item(item_id: int):
if item_id != 1:
return ErrorResponse(code=404, message="Item not found")
return SuccessResponse(data={"id": item_id})
if __name__ == "__main__":
uvicorn.run(app, host="0.0.0.0", port=8000)
time
Timezone-aware time utilities (defaults to Asia/Shanghai).
from yuanfen import time
from datetime import datetime
# Current time (no tzinfo attached)
dt = time.now()
# Current time with timezone info
dt_tz = time.now(tz="Asia/Shanghai", with_tz=True)
# Strip timezone from a tz-aware datetime
dt_naive = time.remove_tz(dt_tz)
# Format datetime to string
time.format(dt=datetime.now(), format="%Y-%m-%dT%H:%M:%S")
# Parse string to datetime
time.parse(dt_str="2023-11-25T10:51:19", format="%Y-%m-%dT%H:%M:%S")
# Format duration in seconds to HH:MM:SS or MM:SS
time.format_duration(90) # "01:30"
time.format_duration(3661) # "01:01:01"
# Timestamps (default 16-digit microseconds)
time.current_timestamp() # current timestamp
time.get_timestamp(dt, length=13) # milliseconds from datetime
# Passthrough to standard library
time.sleep(1.5)
time.time()
GroupRobot
Send messages to a webhook-based group robot (e.g. DingTalk, Feishu).
from yuanfen import GroupRobot
robot = GroupRobot(webhook="https://your-robot-webhook-url")
robot.send({"msgtype": "text", "text": {"content": "Hello!"}})
Send and receive emails via SMTP/IMAP.
from yuanfen import Email
mail = Email(
address="you@example.com",
password="your-password",
smtp_host="smtp.example.com",
smtp_port=465,
imap_host="imap.example.com",
imap_port=993,
sender_name="My App", # optional display name
)
# Send a plain-text email
mail.send_text(to="recipient@example.com", subject="Hello", text="World")
# Search latest N email UIDs matching IMAP criteria
ids = mail.search_ids(count=5, "UNSEEN")
# Fetch a single email by UID
msg = mail.fetch(message_id="123", content_type="text/plain")
# msg keys: subject, from, to, date, charset, content
# Search + fetch in one call
messages = mail.search(count=5, content_type="text/plain", "UNSEEN")
Redis
A Redis client wrapper with optional key prefix and a distributed lock helper.
from yuanfen import Redis, RedisLock
redis = Redis({
"host": "localhost",
"port": 6379,
"password": "secret",
"db": 0,
"prefix": "myapp", # optional; keys are stored as "myapp:<key>"
"decode_responses": True, # optional, default True
})
redis.set("foo", "bar", ex=60) # set with 60s expiry
redis.get("foo") # "bar"
redis.setex("foo", 60, "bar")
redis.setnx("counter", "0")
redis.incr("counter")
redis.delete("foo")
redis.exists("foo")
redis.expire("foo", 120)
redis.ttl("foo")
redis.getset("foo", "new")
RedisLock
lock = RedisLock(
redis_client=redis.redis_client,
lock_key="my-job",
timeout=10, # lock expiry in seconds
retry_interval=0.5, # seconds between retries; None = no retry
)
if lock.acquire():
try:
# critical section
pass
finally:
lock.release()
SMS
SMS verification code service built on Alibaba Cloud SMS (async).
from yuanfen import Config, Logger, SmsService, SmsSendCodeRequest, SmsVerifyCodeRequest
config = Config("config.yaml")
logger = Logger()
sms = SmsService(config=config, logger=logger)
# config.yaml must contain:
# redis: { host, port, password, db, prefix }
# aliyun: { access_key_id, access_key_secret, sms_sign_name, sms_template_code }
# Send a 6-digit code (raises on cooldown or send failure)
await sms.send_code(SmsSendCodeRequest(system="myapp", phone="13800138000", length=6))
# Verify the code (raises on wrong code / expiry / too many attempts)
await sms.verify_code(SmsVerifyCodeRequest(system="myapp", phone="13800138000", code="123456"))
Behavior:
- Code TTL: 300 seconds (5 minutes)
- Resend cooldown: 60 seconds
- Max verify attempts: 5 (code is deleted on exceeded)
hash
from yuanfen import hash
# SHA-256 (default)
digest = hash.get_file_hash("path/to/file")
# Other algorithm, truncated to 8 chars
digest = hash.get_file_hash("path/to/file", algorithm="md5", length=8)
ip
from yuanfen import ip
# Get public IP (tries ipify then ipip.net)
public_ip = ip.get_public_ip()
# Get IP location (sources: "baidu" or "ip-api")
location = ip.get_ip_location("8.8.8.8", source="baidu")
location = ip.get_ip_location("8.8.8.8", source="ip-api")
Version
Semantic version parsing and comparison.
from yuanfen import Version
v = Version(1, 2, 3)
v = Version.parse("1.2.3")
str(v) # "1.2.3"
v.to_tuple() # (1, 2, 3)
v.to_dict() # {"major": 1, "minor": 2, "patch": 3}
# compare() returns -1, 0, or 1
v.compare("1.2.4") # -1
v.compare(Version(1, 2, 3)) # 0
v.compare({"major": 1, "minor": 2, "patch": 0}) # 1
APP_ENV
Reads the APP_ENV environment variable (defaults to "dev").
from yuanfen import APP_ENV
if APP_ENV == "prod":
...
1 maintainer
Metadata
1 maintainer
Release files for yuanfen 2026.3.16.1
For a detailed explanation of source distributions (sdists) and built distributions (wheels), please see the package formats documentation.
Source distribution (sdist)
| File | Size | Uploaded | |
|---|---|---|---|
| yuanfen-2026.3.16.1.tar.gz | 15.2 kB | Details |
Built distribution (wheel)
| File | Interpreter | ABI | Platform | Reset |
|---|---|---|---|---|
| yuanfen-2026.3.16.1-py3-none-any.whl | Python 3 | none | any | Details |
Total release size: 29.7 kB
Release files / yuanfen-2026.3.16.1.tar.gz
| Download URL | yuanfen-2026.3.16.1.tar.gz |
|---|---|
| Size | 15.2 kB |
| Tags | Source |
|
SHA-256 checksum How to use checksums |
7f6c9d923e77e4dfc138d1571246e3f9e1b0e95a03dd88a56c8772baebea893e
|
|
BLAKE2b-256 checksum How to use checksums |
cd97269d5f3e8c31d81575a938e8549e96ab55fce82c52ed802c913e58105c93
|
| Upload date | |
|
Uploaded using Trusted Publishing? What is trusted publishing? |
Yes |
| Uploaded via |
uv/0.10.10 {"installer":{"name":"uv","version":"0.10.10","subcommand":["publish"]},"python":null,"implementation":{"name":null,"version":null},"distro":{"name":"Ubuntu","version":"24.04","id":"noble","libc":null},"system":{"name":null,"release":null},"cpu":null,"openssl_version":null,"setuptools_version":null,"rustc_version":null,"ci":true}
|
Release files / yuanfen-2026.3.16.1-py3-none-any.whl
| Download URL | yuanfen-2026.3.16.1-py3-none-any.whl |
|---|---|
| Size | 14.5 kB |
| Tags | Python 3 |
|
SHA-256 checksum How to use checksums |
1633c953bdfe560989166d704e9bd44f22feb0ebee187632140c97c290312d39
|
|
BLAKE2b-256 checksum How to use checksums |
2a47f653199290d1935b26a552df4f3792f929a269a480ee1ad27b39aade9fb4
|
| Upload date | |
|
Uploaded using Trusted Publishing? What is trusted publishing? |
Yes |
| Uploaded via |
uv/0.10.10 {"installer":{"name":"uv","version":"0.10.10","subcommand":["publish"]},"python":null,"implementation":{"name":null,"version":null},"distro":{"name":"Ubuntu","version":"24.04","id":"noble","libc":null},"system":{"name":null,"release":null},"cpu":null,"openssl_version":null,"setuptools_version":null,"rustc_version":null,"ci":true}
|
