inopyutils
Advanced tools
A comprehensive Python utility library for S3 storage, media processing, file management, configuration, and logging
A comprehensive Python utility library designed for modern development workflows, featuring S3-compatible storage operations, advanced JSON processing, media handling, file management, configuration management, structured logging, an async HTTP client, and audio processing utilities.
⚠️ Active Development
This library is under active development and evolving rapidly. Built to satisfy specific use-cases, APIs may change without prior notice.🔬 Beta Status
Currently in beta stage. While functional, thorough testing is recommended before production use. Please review the code and test extensively for your specific requirements.🤝 Community Welcome
Contributions, feedback, and issue reports are actively encouraged. Help us make this library better for everyone!
InoS3Helper)Universal cloud storage solution supporting AWS S3, Backblaze B2, DigitalOcean Spaces, Wasabi, MinIO, and other S3-compatible services.
Features:
import asyncio
from inopyutils import InoS3Helper
async def main():
# Initialize with Backblaze B2
s3_client = InoS3Helper(
aws_access_key_id='your_key_id',
aws_secret_access_key='your_secret_key',
endpoint_url='https://s3.us-west-004.backblazeb2.com',
region_name='us-west-004',
bucket_name='your-bucket',
retries=5
)
# Async file operations
await s3_client.upload_file('local_file.txt', 'remote/path/file.txt')
await s3_client.download_file('remote/path/file.txt', 'downloaded_file.txt')
# Check file existence and list objects
exists = await s3_client.object_exists('remote/path/file.txt')
listed = await s3_client.list_objects(prefix='remote/path/')
print(exists, listed)
asyncio.run(main())
InoJsonHelper)Comprehensive JSON manipulation toolkit with both synchronous and asynchronous operations, perfect for configuration management and data processing.
Features:
import asyncio
from inopyutils import InoJsonHelper
def demo_sync_ops():
# String/Dict conversions with error handling
result = InoJsonHelper.string_to_dict('{"key": "value"}')
data = result.get("data") if result.get("success") else {}
# Deep operations
dict1 = {"a": 1, "nested": {"x": 10}}
dict2 = {"b": 2, "nested": {"y": 20}}
merged = InoJsonHelper.deep_merge(dict1, dict2)
flattened = InoJsonHelper.flatten({"a": {"b": {"c": 1}}}) # {"a.b.c": 1}
original = InoJsonHelper.unflatten({"a.b.c": 1}) # {"a": {"b": {"c": 1}}}
# Safe path operations
value = InoJsonHelper.safe_get(data, "user.profile.name", default="Unknown")
InoJsonHelper.safe_set(data, "user.profile.age", 25)
# Advanced filtering and searching
cleaned = InoJsonHelper.remove_null_values(data or {}, remove_empty=True)
filtered = InoJsonHelper.filter_keys({"name":"Ann","email":"a@b"}, ["name", "email"], deep=True)
found = InoJsonHelper.find_field_from_array([
{"id": "user_123", "n": 1}, {"id": "user_999", "n": 2}
], "id", "user_123")
# Data comparison with detailed diff
old_data = {"a": 1}
new_data = {"a": 2}
differences = InoJsonHelper.compare(old_data, new_data)
async def demo_async_ops():
await InoJsonHelper.save_json_as_json_async({"config": "data"}, "config.json")
loaded = await InoJsonHelper.read_json_from_file_async("config.json")
print(loaded)
if __name__ == "__main__":
demo_sync_ops()
asyncio.run(demo_async_ops())
InoHttpHelper)High-level asynchronous HTTP client built on aiohttp with robust retry, backoff, timeouts, and base URL support.
Features:
import asyncio
from inopyutils import InoHttpHelper
async def main():
# Create a reusable client with sensible defaults
client = InoHttpHelper(
base_url="https://api.example.com",
timeout_total=30.0,
retries=3,
backoff_factor=0.7,
default_headers={"User-Agent": "InoPyUtils/1.5.2"},
)
# Simple GET returning JSON
resp = await client.get("/users/42", json=True)
# POST JSON and read JSON response
resp = await client.post(
"/items",
json={"name": "Widget", "price": 9.99},
json_response=True,
)
# Download raw bytes
image_bytes = await client.get("/images/logo.png", return_bytes=True)
# Clean up when done (if not using async context manager)
await client.close()
asyncio.run(main())
InoFileHelper)Robust file and folder operations with advanced features for batch processing, archiving, and media validation.
Features:
from inopyutils import InoFileHelper
from pathlib import Path
# Create compressed archives (synchronous)
InoFileHelper.zip(
to_zip=Path("source_folder"),
path_to_save=Path("archives"),
zip_file_name="backup.zip",
compression_level=6,
include_root=False
)
# Batch file operations with smart naming
InoFileHelper.copy_files(
from_path=Path("source"),
to_path=Path("processed"),
rename_files=True,
prefix_name="Processed_",
iterate_subfolders=True
)
# File analysis and utilities
file_count = InoFileHelper.count_files(Path("folder"), recursive=True)
latest_file = InoFileHelper.get_last_file(Path("folder"))
batch_name = InoFileHelper.increment_batch_name("Batch_001") # "Batch_002"
# Media validation and conversion (synchronous)
InoFileHelper.validate_files(
input_path=Path("media_folder"),
include_image=True,
include_video=True,
image_valid_exts=['.jpg', '.png', '.heic'],
video_valid_exts=['.mp4', '.mov']
)
InoMediaHelper)Professional-grade media processing with FFmpeg integration and Pillow-based image manipulation.
Features:
import asyncio
from inopyutils import InoMediaHelper
from pathlib import Path
async def main():
# Advanced image processing
res1 = await InoMediaHelper.image_validate_pillow(
input_path=Path("photo.heic"),
output_path=Path("converted.jpg"),
max_res=2048,
jpg_quality=85,
)
print(res1)
# Video processing with quality control
res2 = await InoMediaHelper.video_convert_ffmpeg(
input_path=Path("input.mov"),
output_path=Path("optimized.mp4"),
change_res=True,
max_res=1920,
change_fps=True,
max_fps=30
)
print(res2)
# Video checks (synchronous helpers)
info = InoMediaHelper.validate_video_res_fps(Path("optimized.mp4"))
fps = InoMediaHelper.get_video_fps(Path("optimized.mp4"))
print(info, fps)
asyncio.run(main())
InoAudioHelper)High-level audio utilities for working with raw PCM streams and common container formats.
Features:
import asyncio
from inopyutils import InoAudioHelper
async def main():
# Load an audio file as bytes (example OGG)
with open("audio.ogg", "rb") as f:
ogg_bytes = f.read()
# Decode audio bytes to raw PCM (s16le, 16kHz, mono)
dec = await InoAudioHelper.audio_to_raw_pcm(
ogg_bytes,
to_format="s16le",
rate=16000,
channel=1,
)
assert dec["success"], dec["error_code"]
pcm_bytes = dec["data"]
# Transcode PCM to OGG/Opus with VOIP application profile
enc = await InoAudioHelper.transcode_raw_pcm(
pcm_bytes,
output="ogg",
codec="libopus",
to_format="s16le",
application="voip",
rate=16000,
channel=1,
)
assert enc["success"], enc["error_code"]
ogg_opus_bytes = enc["data"]
# Stream PCM in fixed-size chunks (e.g., 3200 bytes ~100ms at 16kHz mono s16le)
ch = await InoAudioHelper.chunks_raw_pcm(pcm_bytes, chunk_size=3200)
for chunk in ch["chunks"]:
pass # send chunk to your streaming endpoint
# Estimate TTS duration for pacing
seconds = InoAudioHelper.get_audio_duration_from_text("Hello world", wpm=160.0)
# Produce 2 seconds of silence PCM
silence_pcm = InoAudioHelper.get_empty_audio_pcm_bytes(
duration=2,
to_format="s16le",
rate=16000,
channel=1,
)
asyncio.run(main())
InoThumbnailHelper)Create square thumbnails from images with optional center-crop or smart blurred background padding. Always outputs JPEG files with a consistent naming scheme.
Features:
from pathlib import Path
from inopyutils import InoThumbnailHelper
# Synchronous API
out_files = InoThumbnailHelper.image_generate_square_thumbnails(
image_path=Path("tests/thumbnail_helper/assets/sample.jpg"),
output_dir=Path("tests/thumbnail_helper/thumbnails"),
sizes=(256, 512, 1024),
quality=85,
crop=False, # if True: center-crops to square; if False: pads with blurred background
)
print(out_files) # [".../sample_ino_t_256_.jpg", ".../sample_ino_t_512_.jpg", ...]
# Async API
import asyncio
async def run_async():
out_async = await InoThumbnailHelper.image_generate_square_thumbnails_async(
image_path=Path("photo.heic"),
output_dir=Path("./thumbnails"),
sizes=(256, 768),
quality=80,
crop=True,
)
print(out_async)
asyncio.run(run_async())
InoPhotoMetadata)Lightweight dataclass for holding EXIF-like photo metadata with ready-made profiles.
Features:
from inopyutils import InoPhotoMetadata
# Start from a profile and override what you need
meta = InoPhotoMetadata(profile="iphone")
meta.iso_speed = 100
meta.gps_latitude = 37.7749
meta.gps_longitude = -122.4194
# Use `meta` alongside your own EXIF writing pipeline if needed
print(meta)
InoCsvHelper)Async CSV read/write with convenient in-memory helpers for headers, rows, columns, and sorting.
Features:
import asyncio
from inopyutils import InoCsvHelper
rows = [
{"id": 2, "name": "Bob"},
{"id": 1, "name": "Alice"},
]
async def main():
# Save CSV
res = await InoCsvHelper.save_csv_to_file_async(rows, "people.csv")
assert res["success"], res
# Read CSV
r2 = await InoCsvHelper.read_csv_from_file_async("people.csv")
print(r2["data"]["headers"], len(r2["data"]["rows"]))
# In-memory utilities
headers = InoCsvHelper.get_headers(rows)
first = InoCsvHelper.get_row(rows, 0)
ids = InoCsvHelper.get_column(rows, "id")
sorted_rows = InoCsvHelper.sort_rows(rows, by=["name", "id"]) # multi-key sort
asyncio.run(main())
InoMongoHelper)Typed, high-level async helper around Motor for common MongoDB operations. Initialize once, use everywhere.
import asyncio
from inopyutils import InoMongoHelper
mongo = InoMongoHelper()
async def main():
await mongo.connect(
uri="mongodb://localhost:27017",
db_name="mydb",
serverSelectionTimeoutMS=5_000,
check_connection=True,
)
# CRUD examples
user_id = await mongo.insert_one("users", {"name": "Ann"})
user = await mongo.find_one("users", {"_id": user_id})
await mongo.update_one("users", {"_id": user_id}, {"$set": {"name": "Anna"}})
await mongo.delete_one("users", {"_id": user_id})
await mongo.close()
asyncio.run(main())
Key features:
InoConfigHelper)Robust INI-based configuration management with type safety and debugging capabilities.
Features:
import asyncio
from inopyutils import InoConfigHelper
# Initialize
config = InoConfigHelper('config/application.ini')
# Type-safe configuration access
database_url = config.get('database', 'url', fallback='sqlite:///default.db')
debug_mode = config.get_bool('app', 'debug', fallback=False)
# Configuration updates (sync)
config.set('api', 'endpoint', 'https://api.production.com')
config.save()
# Or async set/save
async def main():
await config.set_async('features', 'cache_enabled', True)
await config.save_async()
asyncio.run(main())
InoLogHelper)Advanced logging system with automatic batching, categorization, and JSON-Lines format output.
Features:
import asyncio
from inopyutils import InoLogHelper, LogType
from pathlib import Path
async def main():
# Initialize logger with automatic batching
logger = await InoLogHelper.create(Path("logs"), "MyApplication")
# Context-rich logging
await logger.info(
msg="User login successful",
log_data={"user_id": 12345, "action": "login", "ip": "192.168.1.100"},
source="auth.login",
)
# Categorized logging with .add
await logger.add(
LogType.ERROR,
msg="API endpoint timeout",
log_data={"error_code": 500, "endpoint": "/api/users", "duration_ms": 1200},
source="api.users",
)
# Batch processing logs
await logger.add(
LogType.INFO,
msg="Batch processing completed",
log_data={"processed": 150, "failed": 3, "batch_id": "batch_20241009"},
source="worker.batch",
)
asyncio.run(main())
pip install inopyutils
# Clone the repository
git clone https://github.com/nobandegani/InoPyUtils.git
cd InoPyUtils
# Install in development mode
pip install -e .
# Install with development dependencies
pip install -e ".[dev]"
# Clone and setup
git clone https://github.com/nobandegani/InoPyUtils.git
cd InoPyUtils
# Create virtual environment
python -m venv venv
source venv/bin/activate # On Windows: venv\Scripts\activate
# Install in development mode
pip install -e ".[dev]"
# Run tests
python -m pytest tests/
git checkout -b feature/amazing-feature)git commit -m 'Add amazing feature')git push origin feature/amazing-feature)This project is licensed under the Mozilla Public License 2.0 (MPL-2.0). See the LICENSE file for details.
Built with ❤️ by the Inoland. Special thanks to all contributors and the open-source community for their invaluable tools and libraries that make this project possible.
FAQs
A comprehensive Python utility library for S3 storage, media processing, file management, configuration, and logging
We found that inopyutils demonstrated a healthy version release cadence and project activity because the last version was released less than a year ago. It has 1 open source maintainer collaborating on the project.
Did you know?
Socket for GitHub automatically highlights issues in each pull request and monitors the health of all your open source dependencies. Discover the contents of your packages and block harmful activity before you install or update your dependencies.
Security News
AI agents are writing more code than ever, and that's creating new supply chain risks. Feross joins the Risky Business Podcast to break down what that means for open source security.
Research
/Security News
Socket uncovered four malicious NuGet packages targeting ASP.NET apps, using a typosquatted dropper and localhost proxy to steal Identity data and backdoor apps.
Research
An emerging npm supply chain attack that infects repos, steals CI secrets, and targets developer AI toolchains for further compromise.