Python 3.14 Support¶

from __future__ import annotations
from logly import logger
import annotationlib

class LogProcessor:
    """Example using deferred annotations with Logly."""

    next_processor: LogProcessor | None = None  # No quotes needed!

    def __init__(self, name: str):
        self.name = name
        logger.info("Processor created", processor_name=name)

    def process(self, message: str) -> LogProcessor | None:
        """Process a log message and return next processor."""
        logger.debug("Processing message", 
                    processor=self.name, 
                    message=message)
        return self.next_processor

# Inspect annotations at runtime
def inspect_processor_annotations():
    annotations = annotationlib.get_annotations(
        LogProcessor, 
        format=annotationlib.Format.VALUE
    )
    logger.info("Processor annotations", annotations=str(annotations))

# Usage
logger.configure(level="DEBUG")
processor = LogProcessor("MainProcessor")
inspect_processor_annotations()
logger.complete()

from logly import logger

# Configure logger
logger.configure(level="INFO")
logger.add("app.log")

# Define reusable t-string templates for log messages
login_template = t"User {username} logged in from {ip}"
error_template = t"Error {code}: {message}"

# Use templates to create log messages
msg1 = login_template.format(username="Alice", ip="192.168.1.1")
logger.info(msg1)

msg2 = error_template.format(code=404, message="Not found")
logger.error(msg2)

from logly import logger

# Logly's format strings define the OUTPUT structure
logger.configure(level="INFO")
logger.add(
    "app.log",
    format="{time:YYYY-MM-DD HH:mm:ss} | {level} | {message}"
    #      ^^^^^^^^^^^^^^^^^^^^^^^^     Logly-specific patterns
)

# Log with custom fields (NOT t-strings)
logger.info("User login", username="Alice", age=30)

from uuid import uuid7
from logly import logger
import time

class RequestHandler:
    """Handle HTTP requests with UUID7 tracking."""

    def handle_request(self, endpoint: str, data: dict):
        # Generate time-sortable UUID
        request_id = uuid7()
        start_time = time.perf_counter()

        # Bind request context
        request_logger = logger.bind(
            request_id=str(request_id),
            endpoint=endpoint
        )

        request_logger.info("Request started", method="POST")

        try:
            # Process request
            result = self._process(data)
            duration_ms = (time.perf_counter() - start_time) * 1000

            request_logger.success("Request completed", 
                                  result=result, 
                                  duration_ms=f"{duration_ms:.2f}")
            return result

        except Exception as e:
            duration_ms = (time.perf_counter() - start_time) * 1000
            request_logger.error("Request failed", 
                               error=str(e),
                               duration_ms=f"{duration_ms:.2f}",
                               exception=True)
            raise

    def _process(self, data: dict) -> str:
        return f"Processed: {data}"

# Usage
logger.configure(level="INFO")
logger.add("requests.log", 
          format="{time:YYYY-MM-DD HH:mm:ss.SSS} | {request_id} | {level} | {message}")

handler = RequestHandler()
handler.handle_request("/api/users", {"user": "alice", "action": "create"})
logger.complete()

from pathlib import Path
from logly import logger
from datetime import datetime

def manage_log_files():
    """Demonstrate pathlib 3.14 features with log files."""
    logger.configure(level="INFO")

    # Create log directory
    log_dir = Path("logs")
    log_dir.mkdir(exist_ok=True)

    # Add log file
    log_file = log_dir / "app.log"
    sink_id = logger.add(str(log_file), rotation="daily")

    logger.info("Application started")
    logger.warning("Low disk space", available_mb=512)

    # Python 3.14: Check file info without separate stat() call
    if log_file.exists():
        info = log_file.info  # New in 3.14!
        logger.info("Log file info",
                   size=info.size,
                   is_file=info.is_file(),
                   modified=datetime.fromtimestamp(info.mtime).isoformat())

    # Python 3.14: Archive logs with new copy() method
    archive_dir = log_dir / "archive"
    archive_dir.mkdir(exist_ok=True)

    if log_file.exists():
        # New copy() method in Python 3.14
        archived = log_file.copy(archive_dir / f"app_{datetime.now():%Y%m%d}.log")
        logger.info("Log archived", archive_path=str(archived))

    logger.complete()

    # Cleanup: use new move() method
    # log_file.move(archive_dir / "final.log")  # New in 3.14

# Usage
manage_log_files()

from concurrent.futures import InterpreterPoolExecutor
from logly import logger
import time

def worker_task(task_id: int) -> dict:
    """Worker function with isolated logging."""
    from logly import logger

    # Each interpreter has its own logger instance
    logger.configure(level="DEBUG")
    logger.add(f"worker_{task_id}.log")

    logger.info("Worker started", task_id=task_id)

    # Simulate work
    time.sleep(0.1)
    result = task_id ** 2

    logger.success("Worker completed", 
                  task_id=task_id, 
                  result=result)

    logger.complete()
    return {"task_id": task_id, "result": result}

def parallel_processing():
    """Demonstrate parallel processing with isolated loggers."""
    logger.configure(level="INFO")
    logger.add("main.log")

    logger.info("Starting parallel processing")

    # Use InterpreterPoolExecutor for true parallelism
    with InterpreterPoolExecutor(max_workers=4) as executor:
        tasks = range(5)
        futures = [executor.submit(worker_task, i) for i in tasks]

        for future in futures:
            result = future.result()
            logger.info("Task completed", **result)

    logger.info("All tasks completed")
    logger.complete()

# Usage
parallel_processing()

from logly import logger

def handle_user_input(data: str):
    """Demonstrate improved exception handling with logging."""
    logger.configure(level="DEBUG")

    try:
        # Attempt to process data
        result = int(data)
        logger.info("Input processed", value=result)

    except ValueError, TypeError:  # New simplified syntax in 3.14!
        # Both exceptions caught without parentheses
        logger.error("Invalid input", 
                    input=data,
                    error_type="ValueError or TypeError")

    except Exception as e:
        # Still works with 'as' clause
        logger.critical("Unexpected error", 
                       error=str(e),
                       exception=True)

    finally:
        logger.debug("Input processing complete")

# Usage
handle_user_input("123")    # Success
handle_user_input("abc")    # ValueError
handle_user_input(None)     # TypeError
logger.complete()

#!/usr/bin/env python3.14
"""Complete demo: Python 3.14 features with Logly logging."""

from __future__ import annotations
from pathlib import Path
from uuid import uuid7
from datetime import datetime
from concurrent.futures import InterpreterPoolExecutor
from logly import logger
import annotationlib

class DataProcessor:
    """Process data with full Python 3.14 + Logly integration."""

    next: DataProcessor | None = None  # PEP 649: no quotes needed

    def __init__(self, name: str):
        self.name = name
        self.request_id = uuid7()  # UUID7 for tracking

        # Bind context
        self.logger = logger.bind(
            processor=name,
            request_id=str(self.request_id)
        )

        self.logger.info("Processor initialized")

    def process(self, data: list[int]) -> dict:
        """Process data with comprehensive logging."""
        self.logger.info("Processing started", items=len(data))

        try:
            # Simulate processing
            result = sum(data)
            self.logger.success("Processing completed", result=result)
            return {"sum": result, "count": len(data)}

        except ValueError, TypeError:  # Python 3.14 syntax
            self.logger.error("Invalid data type")
            return {"error": "Invalid data"}

        except Exception as e:
            self.logger.critical("Unexpected error", 
                               error=str(e),
                               exception=True)
            return {"error": str(e)}

def setup_logging():
    """Configure logging with Python 3.14 pathlib features."""
    logger.configure(level="DEBUG", color=True)

    # Use pathlib to manage log directory
    log_dir = Path("logs")
    log_dir.mkdir(exist_ok=True)

    # Add sinks with time format specs
    logger.add(
        str(log_dir / "app.log"),
        format="{time:YYYY-MM-DD HH:mm:ss.SSS} | {level:8} | {message}",
        rotation="daily"
    )

    logger.add(
        str(log_dir / "requests.log"),
        format="{time:YYYY-MM-DDTHH:mm:ss} | {request_id} | {processor} | {message}",
        filter_min_level="INFO"
    )

    logger.info("Logging configured", log_dir=str(log_dir))

def parallel_processing():
    """Demonstrate InterpreterPoolExecutor with isolated loggers."""
    logger.info("Starting parallel processing")

    def worker(data: list[int]) -> dict:
        from logly import logger
        processor = DataProcessor(f"Worker-{data[0]}")
        return processor.process(data)

    datasets = [[1, 2, 3], [4, 5, 6], [7, 8, 9]]

    with InterpreterPoolExecutor(max_workers=3) as executor:
        futures = [executor.submit(worker, data) for data in datasets]
        results = [f.result() for f in futures]

    logger.info("Parallel processing completed", 
               total_results=len(results))
    return results

def inspect_annotations_demo():
    """Demonstrate PEP 649 annotation inspection."""
    annotations = annotationlib.get_annotations(
        DataProcessor,
        format=annotationlib.Format.VALUE
    )

    logger.debug("Class annotations inspected",
                annotations=str(annotations))

def main():
    """Main entry point."""
    print("=== Python 3.14 + Logly Demo ===\n")

    # Setup
    setup_logging()

    # Demo deferred annotations
    inspect_annotations_demo()

    # Demo single processor
    processor = DataProcessor("MainProcessor")
    result = processor.process([10, 20, 30])
    logger.info("Single processing result", **result)

    # Demo parallel processing
    results = parallel_processing()

    # Cleanup
    logger.complete()
    print("\nDemo complete! Check logs/ directory for output files.")

if __name__ == "__main__":
    main()

# Python 3.14's optimized attribute access helps logger.bind() performance
request_logger = logger.bind(user_id=123, session_id="abc")
request_logger.info("Fast attribute access")  # Faster in 3.14

# Python 3.14's reduced memory overhead for objects
for i in range(10000):
    logger.debug("Memory efficient logging", iteration=i)
    # Uses less memory per log call in 3.14

# Python 3.14 optimizes f-strings (but use lazy evaluation for logs)
value = 42

# ✗ Don't do this (evaluates even if level filtered)
logger.debug(f"Value is {expensive_function()}")

# ✓ Do this (lazy evaluation)
logger.debug("Value is {}", expensive_function())

# Old (Python < 3.14): Required quotes or __future__ import
class OldStyle:
    next: 'OldStyle' = None

# New (Python 3.14): No quotes needed
from __future__ import annotations
class NewStyle:
    next: NewStyle = None  # Works without quotes!

# Old syntax (still works)
try:
    risky_operation()
except (ValueError, TypeError) as e:
    logger.error("Error occurred", error=str(e))

# New syntax (Python 3.14)
try:
    risky_operation()
except ValueError, TypeError as e:  # Simplified!
    logger.error("Error occurred", error=str(e))

#!/usr/bin/env python3.14
"""Simple Python 3.14 + Logly example."""

from __future__ import annotations
from uuid import uuid7
from pathlib import Path
from logly import logger

class UserService:
    """User service with Python 3.14 features."""

    users: dict[str, dict] = {}  # PEP 649: no quotes needed

    def create_user(self, name: str, email: str) -> str:
        """Create user with UUID7 tracking."""
        user_id = str(uuid7())  # Time-sortable UUID

        # Bind context for this operation
        op_logger = logger.bind(
            operation="create_user",
            user_id=user_id,
            user_name=name
        )

        op_logger.info("Creating user")

        try:
            self.users[user_id] = {
                "name": name,
                "email": email,
                "created_at": str(uuid7())
            }

            op_logger.success("User created successfully")
            return user_id

        except ValueError, TypeError:  # Python 3.14 syntax
            op_logger.error("Invalid user data")
            return ""

        except Exception as e:
            op_logger.critical("Failed to create user", 
                             error=str(e),
                             exception=True)
            return ""

def main():
    """Main function."""
    # Configure logging with pathlib
    log_dir = Path("example_logs")
    log_dir.mkdir(exist_ok=True)

    logger.configure(level="INFO", color=True)
    logger.add(
        str(log_dir / "users.log"),
        format="{time:YYYY-MM-DD HH:mm:ss} | {level:8} | {operation} | {user_id} | {message}"
    )

    # Create service and users
    service = UserService()

    user1_id = service.create_user("Alice", "alice@example.com")
    user2_id = service.create_user("Bob", "bob@example.com")
    user3_id = service.create_user("", "invalid")  # This will fail

    print(f"Created users: {user1_id[:8]}..., {user2_id[:8]}...")
    print(f"Failed user: {user3_id}")

    logger.complete()

if __name__ == "__main__":
    main()

# Define a template (notice the 't' prefix)
t = t"Hello, {name}! Welcome to {event}."

# Fill in the placeholders later
message = t.format(name="Gunjan", event="Python Workshop")
print(message)
# Output: Hello, Gunjan! Welcome to Python Workshop.

invite = t"Dear {name}, your seat for {event} is confirmed."

print(invite.format(name="Gunjan", event="Python Workshop"))
print(invite.format(name="Shubham", event="Data Science Seminar"))

#!/usr/bin/env python3.14
"""Using t-strings for reusable log templates with Logly."""

from datetime import datetime
from logly import logger

# Configure Logly
logger.configure(level="INFO")
logger.add("app.log")

# Define reusable log templates
login_template = t"User {user} logged in from IP {ip} at {timestamp}"
purchase_template = t"User {user} purchased {item} for ${amount:.2f}"
error_template = t"Error {code} occurred for user {user}: {details}"

def user_login(user: str, ip: str):
    """Log user login event using template."""
    msg = login_template.format(
        user=user,
        ip=ip,
        timestamp=datetime.now()
    )
    logger.info(msg)

def purchase_event(user: str, amount: float, item: str):
    """Log purchase transaction using template."""
    msg = purchase_template.format(
        user=user,
        item=item,
        amount=amount
    )
    logger.info(msg)

def error_event(user: str, error_code: int, details: str):
    """Log error event using template."""
    msg = error_template.format(
        code=error_code,
        user=user,
        details=details
    )
    logger.error(msg)

# Use the same templates multiple times
user_login("alice", "192.168.1.100")
user_login("bob", "10.0.0.45")

purchase_event("alice", 49.99, "Premium Plan")
purchase_event("charlie", 99.99, "Enterprise License")

error_event("bob", 404, "Resource not found")
error_event("alice", 500, "Database connection failed")

logger.complete()

#!/usr/bin/env python3.14
"""Using t-strings for email notification logging."""

from logly import logger

# Configure Logly
logger.configure(level="INFO")
logger.add("notifications.log")

# Define email templates
email_template = t"""
Hello {name},

Thank you for registering for {event} on {date}.
We look forward to seeing you!

Best regards,
{organizer}
"""

welcome_template = t"Welcome email sent to {email} for user {name}"
reminder_template = t"Reminder sent to {email}: {event} starts in {hours} hours"

def send_welcome_email(name: str, email: str, event: str, date: str, organizer: str):
    """Send welcome email and log it."""
    # Generate email body
    email_body = email_template.format(
        name=name,
        event=event,
        date=date,
        organizer=organizer
    )

    # Log the email (in real app, you'd send it here)
    logger.info(email_body)

    # Log summary
    summary = welcome_template.format(email=email, name=name)
    logger.success(summary)

def send_event_reminder(email: str, event: str, hours: int):
    """Send event reminder and log it."""
    msg = reminder_template.format(
        email=email,
        event=event,
        hours=hours
    )
    logger.info(msg)

# Use templates multiple times
send_welcome_email(
    name="Gunjan",
    email="gunjan@example.com",
    event="Python Bootcamp",
    date="7th Oct 2025",
    organizer="edSlash Team"
)

send_welcome_email(
    name="Shubham",
    email="shubham@example.com",
    event="Data Science Seminar",
    date="15th Oct 2025",
    organizer="edSlash Team"
)

send_event_reminder("gunjan@example.com", "Python Bootcamp", 24)
send_event_reminder("shubham@example.com", "Data Science Seminar", 48)

logger.complete()

#!/usr/bin/env python3.14
"""Using t-strings for application status logging."""

from logly import logger
from datetime import datetime

# Configure Logly
logger.configure(level="INFO")
logger.add("app_status.log")

# Define status templates
startup_template = t"Application {app_name} started at {time} (version {version})"
shutdown_template = t"Application {app_name} stopped at {time} (uptime: {uptime})"
health_template = t"Health check: {service} is {status} (response time: {ms}ms)"

def log_startup(app_name: str, version: str):
    """Log application startup."""
    msg = startup_template.format(
        app_name=app_name,
        time=datetime.now(),
        version=version
    )
    logger.success(msg)

def log_shutdown(app_name: str, uptime: str):
    """Log application shutdown."""
    msg = shutdown_template.format(
        app_name=app_name,
        time=datetime.now(),
        uptime=uptime
    )
    logger.warning(msg)

def log_health_check(service: str, status: str, response_ms: int):
    """Log service health check."""
    msg = health_template.format(
        service=service,
        status=status,
        ms=response_ms
    )

    if status == "healthy":
        logger.success(msg)
    else:
        logger.error(msg)

# Application lifecycle logging
log_startup("UserAPI", "v1.2.3")

log_health_check("Database", "healthy", 45)
log_health_check("Cache", "healthy", 12)
log_health_check("External API", "degraded", 2500)

log_shutdown("UserAPI", "2h 15m 30s")

logger.complete()

#!/usr/bin/env python3.14
"""Simple web API with Python 3.14 + Logly."""

from __future__ import annotations
from uuid import uuid7
from logly import logger
from pathlib import Path

class SimpleAPI:
    """Minimal API handler with UUID7 request tracking."""

    def __init__(self):
        # Setup logging with pathlib
        log_dir = Path("api_logs")
        log_dir.mkdir(exist_ok=True)

        logger.configure(level="INFO")
        logger.add(
            str(log_dir / "requests.log"),
            format="{time:YYYY-MM-DD HH:mm:ss} | {request_id} | {level} | {message}"
        )

    def handle_request(self, method: str, path: str, user_id: int) -> dict:
        """Handle API request with logging."""
        request_id = uuid7()  # Time-sortable ID

        # Bind context
        req_logger = logger.bind(
            request_id=str(request_id)[:8],  # Short ID for readability
            method=method,
            user_id=user_id
        )

        req_logger.info("Request received", path=path)

        try:
            # Simulate processing
            if path == "/users":
                result = {"users": ["Alice", "Bob"]}
            elif path.startswith("/user/"):
                uid = path.split("/")[-1]
                result = {"user_id": uid, "name": "Alice"}
            else:
                raise ValueError("Unknown endpoint")

            req_logger.success("Request completed", status=200)
            return result

        except ValueError, TypeError:  # Python 3.14 simplified syntax
            req_logger.error("Invalid request", status=400)
            return {"error": "Bad request"}

# Usage
api = SimpleAPI()
api.handle_request("GET", "/users", user_id=123)
api.handle_request("GET", "/user/456", user_id=123)
logger.complete()

#!/usr/bin/env python3.14
"""File processor using Python 3.14 pathlib features."""

from pathlib import Path
from logly import logger
from datetime import datetime

class FileProcessor:
    """Process files with enhanced pathlib logging."""

    def __init__(self):
        logger.configure(level="INFO")
        logger.add("processor.log")

    def process_directory(self, dir_path: str) -> dict:
        """Process all text files in directory."""
        data_dir = Path(dir_path)

        if not data_dir.exists():
            logger.error("Directory not found", path=dir_path)
            return {"processed": 0, "errors": 1}

        processed = 0

        for file_path in data_dir.glob("*.txt"):
            # Python 3.14: Use .info property (efficient)
            info = file_path.info

            logger.info("Processing file",
                       file=file_path.name,
                       size_kb=f"{info.size / 1024:.2f}",
                       modified=datetime.fromtimestamp(info.mtime).strftime("%Y-%m-%d"))

            # Process file
            with file_path.open('r') as f:
                line_count = len(f.readlines())

            logger.success("File processed",
                          file=file_path.name,
                          lines=line_count)

            # Python 3.14: Archive with .copy()
            archive_dir = data_dir / "archive"
            archive_dir.mkdir(exist_ok=True)

            archived = file_path.copy(archive_dir / file_path.name)
            logger.debug("File archived", archive=str(archived))

            processed += 1

        logger.info("Processing complete", total_files=processed)
        return {"processed": processed, "errors": 0}

# Usage
processor = FileProcessor()
result = processor.process_directory("data")
print(f"Result: {result}")
logger.complete()

from __future__ import annotations
from uuid import uuid7
from logly import logger
from pathlib import Path
import json

class APIHandler:
    """REST API handler with comprehensive logging."""

    def __init__(self, name: str):
        self.name = name
        self.logger = logger.bind(service=name)

    def handle_request(self, method: str, path: str, data: dict | None = None) -> dict:
        # Generate UUID7 for request tracking
        request_id = uuid7()

        # Bind request context
        request_logger = self.logger.bind(
            request_id=str(request_id),
            method=method,
            path=path
        )

        request_logger.info("Request received", 
                           data_size=len(json.dumps(data or {})))

        try:
            # Process request
            if method == "GET" and path.startswith("/users/"):
                user_id = path.split("/")[-1]
                result = self._get_user(user_id)
            elif method == "POST" and path == "/users":
                result = self._create_user(data)
            else:
                raise ValueError("Invalid endpoint")

            request_logger.success("Request completed", 
                                  status_code=200,
                                  response_size=len(json.dumps(result)))
            return result

        except ValueError, TypeError:  # Python 3.14 syntax
            request_logger.error("Bad request", status_code=400)
            return {"error": "Bad request"}

        except Exception as e:
            request_logger.critical("Internal error", 
                                   status_code=500,
                                   error=str(e),
                                   exception=True)
            return {"error": "Internal server error"}

    def _get_user(self, user_id: str) -> dict:
        # Simulate database lookup
        return {"id": user_id, "name": "Alice", "email": "alice@example.com"}

    def _create_user(self, data: dict) -> dict:
        # Simulate user creation
        user_id = str(uuid7())
        return {"id": user_id, **data}

def setup_api_logging():
    """Configure logging for API service."""
    logger.configure(level="INFO", color=True)

    # Use pathlib for log management
    log_dir = Path("api_logs")
    log_dir.mkdir(exist_ok=True)

    # Main application log
    logger.add(
        str(log_dir / "api.log"),
        format="{time:YYYY-MM-DD HH:mm:ss} | {level:8} | {service} | {request_id} | {message}",
        rotation="1 day",
        retention="7 days"
    )

    # Error log
    logger.add(
        str(log_dir / "errors.log"),
        format="{time:YYYY-MM-DDTHH:mm:ss.SSS} | ERROR | {service} | {request_id} | {message}",
        filter_min_level="ERROR"
    )

# Usage
setup_api_logging()

api = APIHandler("UserService")

# Simulate API calls
print("=== API Request Examples ===")
result1 = api.handle_request("GET", "/users/123")
print(f"GET /users/123: {result1}")

result2 = api.handle_request("POST", "/users", {"name": "Bob", "email": "bob@example.com"})
print(f"POST /users: {result2}")

result3 = api.handle_request("GET", "/invalid")
print(f"GET /invalid: {result3}")

logger.complete()

import asyncio
from __future__ import annotations
from uuid import uuid7
from logly import logger
from pathlib import Path

class AsyncProcessor:
    """Async data processor with Python 3.14 features."""

    def __init__(self, name: str):
        self.name = name
        self.request_id = uuid7()
        self.logger = logger.bind(
            processor=name,
            request_id=str(self.request_id)
        )

    async def process_batch(self, items: list[dict]) -> list[dict]:
        """Process items asynchronously with logging."""
        self.logger.info("Batch processing started", batch_size=len(items))

        tasks = []
        for i, item in enumerate(items):
            task = asyncio.create_task(self._process_item(item, i))
            tasks.append(task)

        try:
            results = await asyncio.gather(*tasks, return_exceptions=True)

            # Handle results and exceptions
            processed_results = []
            for i, result in enumerate(results):
                if isinstance(result, Exception):
                    self.logger.error("Item processing failed", 
                                     item_index=i, 
                                     error=str(result))
                else:
                    processed_results.append(result)

            self.logger.success("Batch processing completed", 
                               processed=len(processed_results),
                               failed=len(results) - len(processed_results))

            return processed_results

        except Exception as e:
            self.logger.critical("Batch processing failed", 
                                error=str(e),
                                exception=True)
            return []

    async def _process_item(self, item: dict, index: int) -> dict:
        """Process individual item."""
        item_logger = self.logger.bind(item_index=index)
        item_logger.debug("Processing item", item_keys=list(item.keys()))

        # Simulate async processing
        await asyncio.sleep(0.01)

        # Simulate random failure
        if index == 2:  # Fail third item
            raise ValueError("Simulated processing error")

        result = {
            "id": item.get("id"),
            "processed": True,
            "timestamp": str(uuid7())
        }

        item_logger.debug("Item processed successfully")
        return result

async def main():
    """Main async function."""
    logger.configure(level="INFO", color=True)

    # Use pathlib for log directory
    log_dir = Path("async_logs")
    log_dir.mkdir(exist_ok=True)

    logger.add(
        str(log_dir / "async.log"),
        format="{time:YYYY-MM-DD HH:mm:ss.SSS} | {level:8} | {processor} | {request_id} | {message}"
    )

    processor = AsyncProcessor("BatchProcessor")

    # Sample data
    items = [
        {"id": "item1", "data": "value1"},
        {"id": "item2", "data": "value2"},
        {"id": "item3", "data": "value3"},  # This will fail
        {"id": "item4", "data": "value4"}
    ]

    results = await processor.process_batch(items)

    print(f"Processed {len(results)} items successfully")

    logger.complete()

# Run async example
print("=== Async Processing Example ===")
asyncio.run(main())

from __future__ import annotations
from pathlib import Path
from uuid import uuid7
from logly import logger
from datetime import datetime
import json

class DataPipeline:
    """Data processing pipeline with Python 3.14 pathlib features."""

    def __init__(self, pipeline_name: str):
        self.pipeline_name = pipeline_name
        self.pipeline_id = uuid7()
        self.logger = logger.bind(
            pipeline=pipeline_name,
            pipeline_id=str(self.pipeline_id)
        )

        # Use pathlib for directory management
        self.data_dir = Path("data")
        self.archive_dir = Path("archive")
        self.data_dir.mkdir(exist_ok=True)
        self.archive_dir.mkdir(exist_ok=True)

    def process_files(self) -> dict:
        """Process all files in data directory."""
        self.logger.info("Pipeline started", data_dir=str(self.data_dir))

        stats = {"processed": 0, "errors": 0, "archived": 0}

        try:
            for file_path in self.data_dir.glob("*.json"):
                try:
                    result = self._process_file(file_path)
                    if result:
                        stats["processed"] += 1
                        self._archive_file(file_path)
                        stats["archived"] += 1
                    else:
                        stats["errors"] += 1

                except Exception as e:
                    self.logger.error("File processing failed",
                                     file=str(file_path),
                                     error=str(e))
                    stats["errors"] += 1

            self.logger.success("Pipeline completed", **stats)
            return stats

        except Exception as e:
            self.logger.critical("Pipeline failed", 
                                error=str(e),
                                exception=True)
            return stats

    def _process_file(self, file_path: Path) -> bool:
        """Process individual file."""
        file_logger = self.logger.bind(file=str(file_path))

        # Use Python 3.14's info property
        info = file_path.info
        file_logger.info("Processing file",
                        size=info.size,
                        modified=datetime.fromtimestamp(info.mtime).isoformat())

        # Read and process JSON
        with file_path.open('r') as f:
            data = json.load(f)

        # Simulate processing
        processed_data = {
            "original": data,
            "processed_at": str(uuid7()),
            "pipeline": self.pipeline_name
        }

        # Write processed data
        output_path = file_path.with_suffix('.processed.json')
        with output_path.open('w') as f:
            json.dump(processed_data, f, indent=2)

        file_logger.success("File processed", output=str(output_path))
        return True

    def _archive_file(self, file_path: Path):
        """Archive processed file using Python 3.14 copy method."""
        timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
        archive_name = f"{file_path.stem}_{timestamp}{file_path.suffix}"
        archive_path = self.archive_dir / archive_name

        # Use Python 3.14's copy method
        copied = file_path.copy(archive_path)
        self.logger.debug("File archived", 
                         original=str(file_path),
                         archive=str(copied))

def setup_pipeline_logging():
    """Configure logging for data pipeline."""
    logger.configure(level="INFO", color=True)

    logger.add(
        "pipeline.log",
        format="{time:YYYY-MM-DD HH:mm:ss} | {level:8} | {pipeline} | {pipeline_id} | {message}",
        rotation="daily"
    )

# Usage
print("=== Data Pipeline Example ===")

# Create sample data files
data_dir = Path("data")
data_dir.mkdir(exist_ok=True)

sample_data = [
    {"id": 1, "name": "Alice", "value": 100},
    {"id": 2, "name": "Bob", "value": 200},
    {"id": 3, "name": "Charlie", "value": 300}
]

for i, data in enumerate(sample_data):
    file_path = data_dir / f"data_{i+1}.json"
    with file_path.open('w') as f:
        json.dump(data, f, indent=2)

setup_pipeline_logging()

pipeline = DataPipeline("DataProcessor")
stats = pipeline.process_files()

print(f"Pipeline stats: {stats}")
logger.complete()

from concurrent.futures import InterpreterPoolExecutor
from __future__ import annotations
from uuid import uuid7
from logly import logger
from pathlib import Path
import time
import random

class Microservice:
    """Microservice with parallel worker processing."""

    def __init__(self, service_name: str):
        self.service_name = service_name
        self.service_id = uuid7()
        self.logger = logger.bind(
            service=service_name,
            service_id=str(self.service_id)
        )

    def handle_requests(self, requests: list[dict]) -> list[dict]:
        """Handle multiple requests in parallel."""
        self.logger.info("Handling requests", count=len(requests))

        # Use InterpreterPoolExecutor for true parallelism
        with InterpreterPoolExecutor(max_workers=4) as executor:
            futures = [executor.submit(self._process_request, req) for req in requests]
            results = [f.result() for f in futures]

        successful = [r for r in results if not r.get("error")]
        failed = [r for r in results if r.get("error")]

        self.logger.success("Requests processed",
                           successful=len(successful),
                           failed=len(failed))

        return results

    def _process_request(self, request: dict) -> dict:
        """Process individual request in isolated interpreter."""
        from logly import logger
        from uuid import uuid7

        # Each worker has its own logger instance
        request_id = uuid7()
        worker_logger = logger.bind(
            request_id=str(request_id),
            worker_id=str(uuid7())
        )

        worker_logger.configure(level="DEBUG")
        worker_logger.add(f"worker_{request_id}.log")

        try:
            worker_logger.info("Worker processing request", 
                              request_type=request.get("type"))

            # Simulate processing time
            time.sleep(random.uniform(0.1, 0.5))

            # Simulate random success/failure
            if random.random() < 0.8:  # 80% success rate
                result = {
                    "request_id": str(request_id),
                    "status": "success",
                    "data": f"Processed {request.get('data', 'unknown')}"
                }
                worker_logger.success("Request completed")
            else:
                raise ValueError("Simulated processing error")

            worker_logger.complete()
            return result

        except ValueError, TypeError:  # Python 3.14 syntax
            worker_logger.error("Processing failed", 
                               error="Invalid request data")
            worker_logger.complete()
            return {
                "request_id": str(request_id),
                "status": "error",
                "error": "Invalid request data"
            }

        except Exception as e:
            worker_logger.critical("Unexpected error", 
                                  error=str(e),
                                  exception=True)
            worker_logger.complete()
            return {
                "request_id": str(request_id),
                "status": "error", 
                "error": str(e)
            }

def setup_microservice_logging():
    """Configure logging for microservice."""
    logger.configure(level="INFO", color=True)

    log_dir = Path("microservice_logs")
    log_dir.mkdir(exist_ok=True)

    logger.add(
        str(log_dir / "service.log"),
        format="{time:YYYY-MM-DD HH:mm:ss} | {level:8} | {service} | {service_id} | {message}",
        rotation="hourly"
    )

# Usage
print("=== Microservice Example ===")

setup_microservice_logging()

service = Microservice("OrderProcessor")

# Sample requests
requests = [
    {"type": "order", "data": "laptop"},
    {"type": "payment", "data": "credit_card"},
    {"type": "shipping", "data": "express"},
    {"type": "order", "data": "mouse"},
    {"type": "payment", "data": "paypal"},
    {"type": "invalid", "data": None}  # This will fail
]

results = service.handle_requests(requests)

print(f"Processed {len(results)} requests")
for result in results:
    status = result.get("status")
    req_id = result.get("request_id", "")[:8]
    print(f"Request {req_id}: {status}")

logger.complete()