Utilities¶
Data storage¶
Regular files¶
Get a path for storing plugin data. Parent directories are created automatically:
model_path = uni.user_data_path("models/voice.onnx")
import requests
response = requests.get(model_url)
model_path.write_bytes(response.content)
Secure JSON files¶
Encrypted file storing a JSON object in /user/data/plugin.{plugin_name}/:
api_credentials = uni.secure_json_file("my_api_credentials")
if api_credentials.exists():
_api_key = api_credentials.read()["api_key"]
api_credentials.save({
"api_key": "secret-key",
"refresh_token": "token-value"
})
Secure JSONL files¶
Encrypted file storing multiple JSON records, optimized for appending:
from datetime import datetime
notes_file = uni.secure_json_lines_file("my_notes")
# Append a record
notes_file.append({
"created": datetime.now().isoformat(),
"message": "Hello, world!",
})
# Read all records
if notes_file.exists():
for note in notes_file.read():
print(f"{note['created']}: {note['message']}")
# Overwrite all records
notes_file.save([])
Direct encryption¶
Encrypt and decrypt strings directly:
encrypted = uni.encrypt("sensitive data")
decrypted = uni.decrypt(encrypted)
Passphrase protection¶
Require a valid passphrase before sensitive operations:
if uni.check_passphrase(user_input):
perform_backup()
Logging¶
logger = uni.get_logger(__name__)
logger.info("Plugin loaded successfully")
logger.debug(f"Processing request: {data}")
logger.warning("API rate limit approaching")
logger.exception("Failed to fetch data: {e!s}")
Parsing helpers¶
HTML to Markdown¶
Convert HTML to clean Markdown for LLM consumption:
html_content = """
<article>
<h1>Breaking News</h1>
<p>Important update about <a href="https://example.com">the topic</a>.</p>
<img src="banner.jpg" alt="Banner">
</article>
"""
# Basic conversion (strips images, preserves links)
markdown = uni.html_to_markdown(html_content)
# Strip links too
markdown = uni.html_to_markdown(html_content, preserve_links=False)
# Truncate long content
markdown = uni.html_to_markdown(html_content, truncate_after=500)
| Parameter | Type | Description |
|---|---|---|
html |
str |
Raw HTML content |
preserve_links |
bool |
Keep URLs in markdown format (default: True) |
truncate_after |
int \| None |
Maximum character length before truncation |
body_width |
int |
Line wrap width, 0 for no wrapping (default: 0) |
Datetime parsing¶
Parse natural language datetime strings:
from datetime import datetime
# Relative times
dt = uni.parse_datetime("tomorrow at 3pm") # -> datetime
dt = uni.parse_datetime("next Tuesday") # -> datetime
dt = uni.parse_datetime("in 2 hours") # -> datetime
# Absolute dates
dt = uni.parse_datetime("December 25th") # -> datetime (next Dec 25)
dt = uni.parse_datetime("2024-03-15 14:30") # -> datetime
# Past dates (disable prefer_future)
dt = uni.parse_datetime("last Monday", prefer_future=False)
| Parameter | Type | Description |
|---|---|---|
text |
str |
Natural language datetime |
prefer_future |
bool |
Interpret ambiguous dates as future (default: True) |
Raises ValueError if the text cannot be parsed.
Duration parsing¶
Parse duration strings into seconds:
# Short format
seconds = uni.parse_duration("2h30m") # -> 9000
seconds = uni.parse_duration("45s") # -> 45
seconds = uni.parse_duration("1h") # -> 3600
# Natural language
seconds = uni.parse_duration("2 hours") # -> 7200
seconds = uni.parse_duration("1 hour 15 minutes") # -> 4500
seconds = uni.parse_duration("90 minutes") # -> 5400
| Parameter | Type | Description |
|---|---|---|
text |
str |
Natural language duration |
Returns duration in seconds as int. Raises ValueError if the text cannot be parsed or is negative.
Caching¶
Cache function results based on arguments:
@uni.cache(ttl="30m")
def fetch_weather_data():
return requests.get("https://api.weather.com/...").json()
@uni.cache(ttl="1h", max_size=100)
def get_user_profile(user_id: str):
return database.query(f"SELECT * FROM users WHERE id = {user_id}")
# Clear cache manually
fetch_weather_data.clear_cache()
| Parameter | Description |
|---|---|
ttl |
Time-to-live (optional, defaults to forever) |
max_size |
Maximum cached entries (optional, unlimited) |
Scheduling¶
One-time tasks¶
from datetime import timedelta
job = uni.schedule_after(timedelta(seconds=30), lambda: print("Runs after 30s"))
job.cancel() # Cancel if needed
Background threads¶
Run functions in managed background threads:
def expensive_computation():
result = process_data()
save_result(result)
uni.run_task(expensive_computation)
Tasks are tracked and waited for during system unload, ensuring clean shutdown. Always use uni.run_task instead of creating raw threads.
For recurring jobs, see Jobs & events.
System information¶
API version¶
Check API version at runtime (semantic versioning):
from packaging.version import Version
MIN_API_VERSION = Version("0.1.0")
if uni.API_VERSION < MIN_API_VERSION:
raise RuntimeError(f"{__name__} requires UNI API version {MIN_API_VERSION}")
CUDA availability¶
if not uni.is_cuda_available():
raise RuntimeError(f"{__name__} requires CUDA")
Hugging Face authentication¶
Check if the user has connected their HF account (required for gated models):
if not uni.is_hf_authenticated():
raise RuntimeError("Please connect UNI to your Hugging Face account.")
For configuration (reading/writing settings), see Getting started. For LLM prompts and embeddings, see Context & prompts.