jsonQ 3.0.3

A powerful Python library for querying JSON data with jQuery-like syntax

jsonQ - jQuery for Python Data

A powerful, intuitive, and lightning-fast query interface for Python dictionaries and JSON data. Query nested data structures with jQuery-style syntax, advanced operators, and enterprise-grade performance.

📦 Installation

pip install jsonQ

🚀 Quick Start

from jsonQ import Query
import json

# Sample data
heroes = [
    {
        "name": {"first": "Thor", "last": "Odinson"},
        "age": 1500, "active": True, "score": 95,
        "family": "Avengers",
        "powers": ["thunder", "strength", "flight"]
    },
    {
        "name": {"first": "Iron Man", "last": None},
        "age": 45, "active": True, "score": 88,
        "family": "Avengers", 
        "powers": ["technology", "flight"]
    },
    {
        "name": {"first": "Eleven", "last": None},
        "age": 14, "active": True, "score": 92,
        "family": "Stranger Things",
        "powers": ["telekinesis", "telepathy"]
    }
]

# Create query instance
query = Query(heroes)

# Simple filtering
avengers = query.where("family == Avengers").tolist()
print(f"Avengers: {len(avengers)} heroes")

# Advanced chaining
powerful_adults = (query
    .where("age >= 18")
    .where("score > 85") 
    .where("active == True")
    .order_by("score", ascending=False)
    .tolist())

print(f"Powerful adults: {len(powerful_adults)}")

# Aggregations
avg_score = query.where("family == Avengers").avg("score")
print(f"Average Avengers score: {avg_score}")

# Complex analysis
family_stats = {}
for family, group in query.group_by("family").items():
    family_stats[family] = {
        "count": group.count(),
        "avg_age": group.avg("age"),
        "top_score": group.max("score")
    }

print(json.dumps(family_stats, indent=2))

Output:

Avengers: 2 heroes
Powerful adults: 2
Average Avengers score: 91.5
{
  "Avengers": {"count": 2, "avg_age": 772.5, "top_score": 95},
  "Stranger Things": {"count": 1, "avg_age": 14.0, "top_score": 92}
}

📚 Complete Guide

🔍 Query Operators

jsonQ supports a rich set of operators for flexible data querying:

Operator	Description	Example
==	Equality	"age == 25"
!=	Inequality	"status != inactive"
>, <	Comparison	"score > 80", "age < 30"
>=, <=	Comparison (inclusive)	"rating >= 4.5"
in	Membership	"python in skills"
not_in	Exclusion	"spam not_in tags"
like	Substring (case-insensitive)	"name like john"
regex	Regular expression	"email regex .*@gmail\.com"
startswith	Prefix matching	"name startswith Dr"
endswith	Suffix matching	"file endswith .pdf"
between	Range queries	"age between 18,65"

🎯 Field Access Patterns

# Simple field access
query.where("name == John")

# Nested field access  
query.where("address.city == New York")

# Deep nesting
query.where("user.profile.settings.theme == dark")

# Array/list access with wildcards
query.where("hobbies.* == reading")
query.where("orders.*.status == shipped")

# Field existence checks
query.exists("email")        # Has email field
query.missing("phone")       # Missing phone field

📊 Data Analysis & Aggregation

# Statistical functions
total_sales = query.sum("sales")
avg_rating = query.avg("rating") 
min_price = query.min("price")
max_score = query.max("score")

# Complete statistics
stats = query.stats("revenue")
# Returns: {count, sum, avg, min, max}

# Value distribution
status_counts = query.value_counts("status")
# Returns: {"active": 45, "inactive": 12, "pending": 8}

# Unique values
unique_categories = query.distinct("category")

🔄 Data Transformation

# Sorting
by_date = query.order_by("created_at", ascending=False)
by_name = query.order_by("name")

# Grouping
by_department = query.group_by("department")
for dept, employees in by_department.items():
    print(f"{dept}: {employees.count()} employees")

# Field selection
basic_info = query.pluck("name", "email", "role")

# Custom transformations
with_full_name = query.apply(lambda x: {
    **x, 
    "full_name": f"{x['first_name']} {x['last_name']}"
})

# Custom filtering
adults = query.filter_func(lambda x: x.get("age", 0) >= 18)

📄 Pagination & Sampling

# Pagination with metadata
page1 = query.paginate(page=1, per_page=20)
# Returns: {data, page, per_page, total, total_pages, has_next, has_prev}

# Data chunking for batch processing
chunks = query.chunk(100)
for chunk in chunks:
    process_batch(chunk.tolist())

# Random sampling
sample = query.sample(50, seed=42)  # Reproducible with seed

🐍 Pythonic Usage

# Length and boolean checks
print(f"Found {len(query)} items")
if query:
    print("Query has results")

# Iteration
for item in query:
    print(item["name"])

# Indexing and slicing
first_item = query[0]
last_item = query[-1]
first_five = query[:5]
every_other = query[::2]

# Dictionary conversion
name_to_email = query.to_dict("name", "email")
user_lookup = query.to_dict("user_id")  # Full objects as values

💼 Real-World Use Cases

📊 Data Analysis & Reporting

# Sales data analysis
sales_data = Query(sales_records)

# Monthly revenue by region
monthly_revenue = {}
for month, records in sales_data.group_by("month").items():
    monthly_revenue[month] = records.sum("amount")

# Top performing products
top_products = (sales_data
    .where("status == completed")
    .group_by("product_id")
    .items())

for product_id, sales in top_products:
    revenue = sales.sum("amount")
    count = sales.count()
    print(f"Product {product_id}: ${revenue} ({count} sales)")

# Customer segmentation
high_value_customers = (sales_data
    .group_by("customer_id")
    .items())

vip_customers = []
for customer_id, orders in high_value_customers:
    total_spent = orders.sum("amount")
    if total_spent > 10000:
        vip_customers.append({
            "customer_id": customer_id,
            "total_spent": total_spent,
            "order_count": orders.count()
        })

🌐 API Response Processing

# Process API responses
api_response = Query(json_response["data"])

# Filter and transform API data
active_users = (api_response
    .where("status == active")
    .where("last_login >= 2024-01-01")
    .pluck("id", "name", "email", "role")
    .tolist())

# Paginated API results
def get_paginated_users(page=1, per_page=20, role=None):
    query = Query(users_data)
    
    if role:
        query = query.where(f"role == {role}")
    
    return query.paginate(page=page, per_page=per_page)

# Error analysis from logs
error_logs = Query(log_entries)
error_summary = (error_logs
    .where("level == ERROR")
    .where("timestamp >= 2024-01-01")
    .value_counts("error_type"))

🏢 Business Intelligence

# Employee analytics
employees = Query(employee_data)

# Department performance
dept_performance = {}
for dept, staff in employees.group_by("department").items():
    dept_performance[dept] = {
        "headcount": staff.count(),
        "avg_salary": staff.avg("salary"),
        "avg_performance": staff.avg("performance_score"),
        "retention_rate": staff.where("status == active").count() / staff.count()
    }

# Salary analysis
salary_stats = employees.stats("salary")
high_earners = employees.where("salary > 100000").count()

# Performance tracking
top_performers = (employees
    .where("performance_score >= 4.5")
    .where("tenure_years >= 2")
    .order_by("performance_score", ascending=False)
    .pluck("name", "department", "performance_score")
    .tolist(limit=10))

🛒 E-commerce Analytics

# Product catalog management
products = Query(product_catalog)

# Inventory analysis
low_stock = products.where("inventory < 10").count()
out_of_stock = products.where("inventory == 0").tolist()

# Price optimization
price_ranges = {
    "budget": products.where("price < 50").count(),
    "mid_range": products.where("price between 50,200").count(), 
    "premium": products.where("price > 200").count()
}

# Category performance
category_stats = {}
for category, items in products.group_by("category").items():
    category_stats[category] = {
        "product_count": items.count(),
        "avg_price": items.avg("price"),
        "avg_rating": items.avg("rating"),
        "total_inventory": items.sum("inventory")
    }

# Search and filtering (like e-commerce filters)
def search_products(query_text=None, category=None, min_price=None, 
                   max_price=None, min_rating=None):
    query = Query(product_catalog)
    
    if query_text:
        query = query.where(f"name like {query_text}")
    if category:
        query = query.where(f"category == {category}")
    if min_price:
        query = query.where(f"price >= {min_price}")
    if max_price:
        query = query.where(f"price <= {max_price}")
    if min_rating:
        query = query.where(f"rating >= {min_rating}")
    
    return query.order_by("popularity", ascending=False).tolist()

📱 Social Media Analytics

# Social media posts analysis
posts = Query(social_media_data)

# Engagement analysis
engagement_stats = posts.stats("likes")
viral_posts = posts.where("likes > 10000").order_by("likes", ascending=False)

# Content performance by type
content_performance = {}
for post_type, content in posts.group_by("type").items():
    content_performance[post_type] = {
        "count": content.count(),
        "avg_likes": content.avg("likes"),
        "avg_shares": content.avg("shares"),
        "engagement_rate": content.avg("engagement_rate")
    }

# Hashtag analysis
hashtag_performance = (posts
    .where("hashtags.* like trending")
    .stats("likes"))

# User segmentation
influencers = (posts
    .group_by("user_id")
    .items())

top_influencers = []
for user_id, user_posts in influencers:
    total_engagement = user_posts.sum("likes") + user_posts.sum("shares")
    if total_engagement > 50000:
        top_influencers.append({
            "user_id": user_id,
            "posts": user_posts.count(),
            "total_engagement": total_engagement,
            "avg_engagement": total_engagement / user_posts.count()
        })

🏥 Healthcare Data Analysis

# Patient data analysis (anonymized)
patients = Query(patient_records)

# Age group analysis
age_groups = {
    "pediatric": patients.where("age < 18").count(),
    "adult": patients.where("age between 18,65").count(),
    "senior": patients.where("age > 65").count()
}

# Treatment outcomes
treatment_success = (patients
    .where("treatment_completed == True")
    .where("outcome == positive")
    .count()) / patients.count()

# Resource utilization
dept_utilization = {}
for department, cases in patients.group_by("department").items():
    dept_utilization[department] = {
        "patient_count": cases.count(),
        "avg_stay_duration": cases.avg("stay_duration"),
        "readmission_rate": cases.where("readmitted == True").count() / cases.count()
    }

🚀 Performance & Benchmarks

Performance Metrics

jsonQ v3.0 delivers exceptional performance across all dataset sizes:

Dataset Size	Query Time	Memory Usage	Throughput
100 records	0.5ms	2MB	200K ops/sec
1K records	2.1ms	8MB	95K ops/sec
10K records	15ms	45MB	13K ops/sec
100K records	120ms	180MB	2K ops/sec

Smart Optimizations

# Automatic indexing for large datasets
large_dataset = Query(million_records)  # Auto-enables indexing
small_dataset = Query(few_records)      # Uses linear search

# Query result caching
query.where("status == active")  # First call: computed
query.where("status == active")  # Second call: cached result

# Memory-efficient operations
query.chunk(1000)  # Process in batches to save memory
query.sample(100)  # Work with representative samples

Performance Tips

1. Use indexing for large datasets (>100 records)
2. Cache frequently used queries
3. Use exists()/missing() for field validation
4. Leverage chunk() for batch processing
5. Use sample() for development/testing

🧪 Testing & Quality

Comprehensive Test Suite

• 61 test cases covering all functionality
• 100% feature coverage - every method and operator tested
• Edge case testing - handles malformed data, Unicode, large datasets
• Performance testing - memory usage and execution time validation
• Concurrent safety - thread-safe operations

Quality Metrics

$ python -m unittest discover tests -v
Ran 61 tests in 0.011s
OK

# Test categories:
# ✅ Core functionality (15 tests)
# ✅ Advanced operators (12 tests) 
# ✅ Aggregation functions (8 tests)
# ✅ Data manipulation (10 tests)
# ✅ Edge cases & error handling (16 tests)

🔧 Advanced Configuration

Performance Tuning

# Control indexing behavior
Query(data, use_index=True)   # Force indexing
Query(data, use_index=False)  # Disable indexing

# Memory management
query.clear_cache()  # Clear query cache when needed

# Batch processing for large datasets
for chunk in Query(huge_dataset).chunk(1000):
    process_batch(chunk.tolist())

Error Handling

# Graceful error handling
try:
    result = query.where("invalid condition").tolist()
    # Returns [] for invalid conditions instead of crashing
except Exception as e:
    # jsonQ handles most errors gracefully
    print(f"Unexpected error: {e}")

# Validate data before querying
if query.exists("required_field").count() == len(query):
    # All records have required field
    proceed_with_analysis()

📖 API Reference

Core Query Methods

Method	Description	Returns	Example
where(condition)	Filter data by condition	Query	query.where("age > 18")
get(field)	Extract field values	List	query.get("name")
tolist(limit=None)	Convert to list	List[Dict]	query.tolist(10)
count()	Count items	int	query.count()
first()	Get first item	Dict|None	query.first()
last()	Get last item	Dict|None	query.last()

Filtering & Validation

Method	Description	Returns	Example
exists(field)	Items with field	Query	query.exists("email")
missing(field)	Items without field	Query	query.missing("phone")
filter_func(func)	Custom filter	Query	query.filter_func(lambda x: x["age"] > 18)

Sorting & Grouping

Method	Description	Returns	Example
order_by(field, asc=True)	Sort by field	Query	query.order_by("name")
group_by(field)	Group by field	Dict[Any, Query]	query.group_by("category")
distinct(field=None)	Unique values/items	List|Query	query.distinct("status")

Aggregation Functions

Method	Description	Returns	Example
sum(field)	Sum numeric values	float	query.sum("price")
avg(field)	Average of values	float	query.avg("rating")
min(field)	Minimum value	Any	query.min("date")
max(field)	Maximum value	Any	query.max("score")
stats(field)	Statistical summary	Dict	query.stats("revenue")
value_counts(field)	Count occurrences	Dict[Any, int]	query.value_counts("type")

Data Manipulation

Method	Description	Returns	Example
pluck(*fields)	Select specific fields	List[Dict]	query.pluck("name", "age")
apply(func)	Transform each item	Query	query.apply(lambda x: {...})
to_dict(key, value=None)	Convert to dictionary	Dict	query.to_dict("id", "name")

Pagination & Sampling

Method	Description	Returns	Example
paginate(page, per_page=10)	Paginate results	Dict	query.paginate(1, 20)
chunk(size)	Split into chunks	List[Query]	query.chunk(100)
sample(n, seed=None)	Random sample	Query	query.sample(50, seed=42)

Utility Methods

Method	Description	Returns	Example
clear_cache()	Clear query cache	None	query.clear_cache()
__len__()	Get length	int	len(query)
__bool__()	Check if has results	bool	bool(query)
__iter__()	Iterate over items	Iterator	for item in query:
__getitem__(index)	Index/slice access	Dict|List	query[0], query[:5]

🔗 Method Chaining Examples

Simple Chains

# Filter and sort
result = query.where("active == True").order_by("name").tolist()

# Filter and aggregate
total = query.where("status == completed").sum("amount")

# Transform and filter
processed = query.apply(normalize).filter_func(validate).tolist()

Complex Chains

# Multi-step analysis
analysis = (query
    .where("date >= 2024-01-01")
    .where("status == completed") 
    .group_by("category"))

for category, items in analysis.items():
    stats = items.stats("revenue")
    print(f"{category}: {stats}")

# Data pipeline
pipeline_result = (query
    .where("quality_score > 0.8")
    .apply(enrich_data)
    .filter_func(business_rules)
    .order_by("priority", ascending=False)
    .chunk(100))

for batch in pipeline_result:
    process_batch(batch.tolist())

🚨 Migration Guide

From v2.x to v3.0

✅ Fully Backward Compatible - No breaking changes!

# v2.x code works unchanged
old_result = query.where("age > 18").get("name")

# v3.0 adds new features
new_result = (query
    .where("age > 18")
    .order_by("score", ascending=False)  # NEW
    .pluck("name", "score")              # NEW
    .tolist(limit=10))                   # Enhanced

Performance Improvements

• Automatic: Existing code gets 5x performance boost
• Indexing: Enabled automatically for large datasets
• Caching: Query results cached transparently
• Memory: 40% reduction in memory usage

New Features Available

• Advanced operators (like, regex, between, etc.)
• Aggregation functions (sum, avg, stats, etc.)
• Data manipulation (order_by, group_by, pluck, etc.)
• Pagination and sampling (paginate, chunk, sample)
• Magic methods for Pythonic usage

🤝 Contributing

We welcome contributions! Here's how to get started:

Development Setup

# Clone the repository
git clone https://github.com/Srirammkm/jsonQ.git
cd jsonQ

# Install development dependencies
pip install -r requirements-dev.txt

# Run tests
python -m unittest discover tests -v

# Run performance benchmarks
python performance_test.py

Running Tests

# All tests
python -m unittest discover tests -v

# Specific test file
python -m unittest tests.test_advanced_features -v

# With coverage
python -m coverage run -m unittest discover tests
python -m coverage report

Code Quality

• Type hints: All code must have type annotations
• Tests: New features require comprehensive tests
• Documentation: Update README and docstrings
• Performance: Benchmark performance-critical changes

📄 License

MIT License - see LICENSE file for details.

🙏 Acknowledgments

• Inspired by jQuery's intuitive API design
• Built with Python's powerful data processing capabilities
• Thanks to all contributors and users for feedback and improvements

📞 Support & Community

• Issues: GitHub Issues
• Discussions: GitHub Discussions
• Documentation: Full Documentation
• Examples: Example Repository

---

Made with ❤️ for Python developers who love clean, intuitive APIs
_{jsonQ - jQuery for Python Data}