Tutorials

Step-by-step guides and examples for using PyWizardry

File Operations Tutorial

Learn how to work with files and directories using PyWizardry

Step 1: Finding Files

from pywizardry import FileMagic

# Find all Python files recursively
files = FileMagic.find_files(
    "/projects",
    pattern="*.py",
    recursive=True
)

# Find files modified in last 7 days
from datetime import datetime, timedelta
recent_files = FileMagic.find_files(
    "/logs",
    modified_after=datetime.now() - timedelta(days=7)
)

Step 2: Safe File Operations

# Safe write with backup
FileMagic.safe_write(
    "important_data.json",
    '{"key": "value"}',
    backup=True,
    atomic=True
)

# Read large files efficiently
for chunk in FileMagic.read_large_file("large.log", chunk_size=8192):
    process_chunk(chunk)

Step 3: File Metadata

# Get comprehensive file info
metadata = FileMagic.get_file_info("data.csv")

print(f"Size: {metadata['size_human']}")
print(f"Created: {metadata['created']}")
print(f"Hash: {metadata['hash_sha256']}")
print(f"MIME Type: {metadata['mime_type']}")

# Check if file exists and is readable
if metadata['is_file']:
    print("File is accessible")

String Manipulation Tutorial

Master string operations with PyWizardry's StringSorcery

Step 1: Basic String Operations

from pywizardry import StringSorcery

# Slugify text for URLs
slug = StringSorcery.slugify("My Awesome Blog Post!")
# Returns: "my-awesome-blog-post"

# Smart truncation
truncated = StringSorcery.truncate(
    "This is a very long text that needs to be shortened",
    max_length=30,
    preserve_words=True
)
# Returns: "This is a very long text..."

Step 2: Password Generation

# Generate secure password
password = StringSorcery.generate_password(
    length=16,
    include_uppercase=True,
    include_digits=True,
    include_symbols=True,
    exclude_similar=True
)

# Generate multiple passwords
passwords = [
    StringSorcery.generate_password() 
    for _ in range(5)
]

Step 3: Text Extraction

# Extract information from text
text = """
Contact us at support@example.com
Visit https://example.com
Call +1-234-567-8900
Follow @username on Twitter
"""

info = StringSorcery.extract_info(text)
# Returns dictionary with emails, URLs, phone numbers, mentions

# Extract just emails
emails = StringSorcery.extract_emails(text)
# Returns: ["support@example.com"]

Security Utilities Tutorial

Implement security features with PyWizardry

Step 1: Password Hashing

from pywizardry import SecuritySpells

# Hash password with salt
hashed = SecuritySpells.hash_password(
    "my_secure_password",
    algorithm="pbkdf2_sha256"
)

# Store in database: hashed['hash'] and hashed['salt']

# Verify password
is_valid = SecuritySpells.verify_password(
    "my_secure_password",
    hashed
)

Step 2: JWT Tokens

# Create JWT token
token = SecuritySpells.jwt_encode(
    {
        "user_id": 123,
        "role": "admin",
        "email": "user@example.com"
    },
    "your-secret-key",
    expires_in=3600  # 1 hour
)

# Verify and decode token
try:
    payload = SecuritySpells.jwt_decode(token, "your-secret-key")
    print(f"User ID: {payload['user_id']}")
except Exception as e:
    print(f"Invalid token: {e}")

Step 3: Input Sanitization

# Sanitize user input
user_input = 'Hello World'
safe_input = SecuritySpells.sanitize_input(user_input)
# Returns: "Hello World"

# Sanitize with allowed tags
html_input = 'Hello World'
safe_html = SecuritySpells.sanitize_input(
    html_input,
    allowed_tags=['p', 'strong', 'em']
)
# Returns: 'Hello World
'

Async Programming Tutorial

Master asynchronous programming with PyWizardry

Step 1: Basic Async Operations

import asyncio
from pywizardry import AsyncMagic

# Run coroutine with timeout
async def fetch_data():
    data = await AsyncMagic.run_with_timeout(
        expensive_operation(),
        timeout=5.0,
        default=None
    )
    return data

# Convert sync function to async
sync_func = lambda: "Hello World"
async_func = AsyncMagic.to_async(sync_func)

result = await async_func()

Step 2: Parallel Processing

from pywizardry import DataAlchemy

# Process data in parallel
def process_item(item):
    # Some CPU-intensive task
    return item * 2

items = list(range(1000))
results = DataAlchemy.batch_process(
    items,
    process_func=process_item,
    batch_size=100,
    max_workers=4
)

Step 3: Async Network Requests

from pywizardry import NetworkEnchantments

async def fetch_multiple_urls(urls):
    tasks = []
    for url in urls:
        task = NetworkEnchantments.async_fetch_json(url)
        tasks.append(task)
    
    # Run all requests concurrently
    results = await asyncio.gather(*tasks)
    return results

# Usage
urls = [
    "https://api.example.com/data1",
    "https://api.example.com/data2",
    "https://api.example.com/data3"
]

results = await fetch_multiple_urls(urls)

Data Processing Tutorial

Process and analyze data efficiently

Step 1: Data Transformation

from pywizardry import DataAlchemy

# Chunk data for processing
data = list(range(1000))
chunks = DataAlchemy.chunk_data(data, chunk_size=100)

# Remove duplicates while preserving order
unique_data = DataAlchemy.remove_duplicates(
    [1, 2, 2, 3, 4, 4, 5],
    key=lambda x: x
)

# Group data by category
items = [
    {"category": "A", "value": 1},
    {"category": "B", "value": 2},
    {"category": "A", "value": 3}
]

grouped = DataAlchemy.group_by(
    items,
    key_func=lambda x: x["category"]
)

Step 2: Statistical Analysis

# Calculate statistics
data = [10, 20, 30, 40, 50]
stats = DataAlchemy.calculate_statistics(data)

print(f"Mean: {stats['mean']}")
print(f"Median: {stats['median']}")
print(f"Standard Deviation: {stats['std']}")
print(f"95th Percentile: {stats['p95']}")

# Detect outliers
outliers = DataAlchemy.detect_outliers(
    data,
    method="iqr",
    threshold=1.5
)

Step 3: Data Normalization

# Normalize data to range [0, 1]
data = [10, 20, 30, 40, 50]
normalized = DataAlchemy.normalize(data)

# Standardize data (z-score)
standardized = DataAlchemy.standardize(data)

# Calculate moving average
prices = [100, 102, 101, 105, 103, 106]
ma = DataAlchemy.moving_average(prices, window_size=3)

# Interpolate missing values
data_with_nulls = [1, None, 3, None, 5, None, 7]
interpolated = DataAlchemy.interpolate_missing(
    data_with_nulls,
    method="linear"
)

Practice Exercises

Exercise 1: File Processor

Create a script that:

Finds all CSV files in a directory
Processes each file and extracts statistics
Saves results to a JSON file with backup
Logs the process with timestamps

from pywizardry import Wizard, FileMagic
import json
from datetime import datetime

wiz = Wizard()

def process_csv_files(directory):
    # Find CSV files
    csv_files = FileMagic.find_files(
        directory,
        pattern="*.csv",
        recursive=True
    )
    
    results = []
    for file_path in csv_files:
        # Get file info
        info = FileMagic.get_file_info(file_path)
        
        # Read and process file (simplified)
        with open(file_path, 'r') as f:
            lines = f.readlines()
        
        # Calculate simple statistics
        stats = {
            "filename": str(file_path.name),
            "size": info["size"],
            "lines": len(lines),
            "processed_at": datetime.now().isoformat()
        }
        
        results.append(stats)
    
    # Save results
    output_data = {
        "processed_at": datetime.now().isoformat(),
        "total_files": len(csv_files),
        "results": results
    }
    
    FileMagic.safe_write(
        "processing_results.json",
        json.dumps(output_data, indent=2),
        backup=True
    )
    
    return results

# Run the processor
process_csv_files("/data")

Exercise 2: API Client with Retry Logic

Create an API client that:

Fetches data from multiple endpoints
Implements retry logic for failed requests
Handles rate limiting
Caches responses

from pywizardry import (
    Wizard, 
    retry_spell, 
    cache_spell,
    spell_timer
)
import asyncio

wiz = Wizard()

@spell_timer()
@retry_spell(max_attempts=3, delay=2.0)
@cache_spell(ttl=300)  # Cache for 5 minutes
def fetch_api_data(url):
    response = wiz.network.fetch_json(url)
    if response["status"] == 200:
        return response["data"]
    else:
        raise Exception(f"API Error: {response.get('error', 'Unknown')}")

class APIClient:
    def __init__(self, base_url):
        self.base_url = base_url
    
    def get_user(self, user_id):
        url = f"{self.base_url}/users/{user_id}"
        return fetch_api_data(url)
    
    def get_posts(self, user_id):
        url = f"{self.base_url}/users/{user_id}/posts"
        return fetch_api_data(url)
    
    async def get_all_data(self, user_ids):
        tasks = []
        for user_id in user_ids:
            # Fetch user and posts concurrently
            user_task = asyncio.create_task(
                self._async_fetch(f"{self.base_url}/users/{user_id}")
            )
            posts_task = asyncio.create_task(
                self._async_fetch(f"{self.base_url}/users/{user_id}/posts")
            )
            tasks.extend([user_task, posts_task])
        
        results = await asyncio.gather(*tasks, return_exceptions=True)
        return results
    
    async def _async_fetch(self, url):
        return await wiz.network.async_fetch_json(url)

# Usage
client = APIClient("https://api.example.com")
user_data = client.get_user(123)
print(user_data)

Ready for More?

API Reference

Explore all available functions

Documentation

Detailed usage guides

Contribute

Help improve PyWizardry