Python for AI: Beginner to Advanced

Data / Input
    ↓
 Python Code
    ↓
 AI Logic / Model
    ↓
 Prediction / Output

name = "Nisha"
print("Hello", name)

def add(a, b):
    return a + b

print(add(5, 3))

import requests
import json

# Fetch data from a public API (JSONPlaceholder is a free test API)
def fetch_user_posts(user_id: int) -> list:
    """Fetches posts for a given user from a REST API."""
    url = f"https://jsonplaceholder.typicode.com/posts"
    
    try:
        # Make GET request with query parameters
        response = requests.get(url, params={"userId": user_id}, timeout=10)
        
        # Always check status code before using data
        response.raise_for_status()  # raises error for 4xx/5xx
        
        posts = response.json()
        return posts
        
    except requests.exceptions.ConnectionError:
        print("Error: Could not connect to API")
        return []
    except requests.exceptions.Timeout:
        print("Error: Request timed out")
        return []
    except requests.exceptions.HTTPError as e:
        print(f"HTTP Error: {e}")
        return []

# Process the results
posts = fetch_user_posts(user_id=1)

print(f"Found {len(posts)} posts for user 1")
for post in posts[:3]:  # Show first 3
    print(f"  - {post['title'][:50]}")

# POST request example: send data to an API
def create_post(title: str, body: str, user_id: int) -> dict:
    """Creates a new post via POST request."""
    url = "https://jsonplaceholder.typicode.com/posts"
    
    payload = {
        "title": title,
        "body": body,
        "userId": user_id
    }
    
    response = requests.post(url, json=payload, timeout=10)
    response.raise_for_status()
    return response.json()

new_post = create_post("My Study Notes", "Today I learned about Python APIs.", user_id=1)
print("\nCreated post with ID:", new_post.get("id"))

import json
import csv
from pathlib import Path

# ── JSON: Read and Write ──

def save_results(data: list, filepath: str):
    """Saves a list of dictionaries to a JSON file."""
    with open(filepath, "w", encoding="utf-8") as f:
        json.dump(data, f, indent=2, ensure_ascii=False)
    print(f"Saved {len(data)} records to {filepath}")

def load_results(filepath: str) -> list:
    """Loads JSON data from a file. Returns empty list if file not found."""
    path = Path(filepath)
    if not path.exists():
        print(f"File not found: {filepath}")
        return []
    with open(filepath, "r", encoding="utf-8") as f:
        return json.load(f)

# Example usage
student_data = [
    {"name": "Arjun", "score": 85, "subject": "Math"},
    {"name": "Priya", "score": 92, "subject": "Science"},
    {"name": "Ravi",  "score": 78, "subject": "English"},
]

save_results(student_data, "students.json")
loaded = load_results("students.json")
print("Loaded:", loaded[0])

# ── CSV: Read and Write ──

def save_to_csv(data: list, filepath: str):
    """Saves a list of dicts to CSV."""
    if not data:
        return
    with open(filepath, "w", newline="", encoding="utf-8") as f:
        writer = csv.DictWriter(f, fieldnames=data[0].keys())
        writer.writeheader()
        writer.writerows(data)
    print(f"CSV saved: {filepath}")

def read_csv(filepath: str) -> list:
    """Reads a CSV into a list of dicts."""
    with open(filepath, "r", encoding="utf-8") as f:
        return list(csv.DictReader(f))

save_to_csv(student_data, "students.csv")
rows = read_csv("students.csv")
print("CSV row 1:", rows[0])

# ── Text files: Append logging ──

def log_event(message: str, logfile: str = "app.log"):
    """Appends a log message with timestamp."""
    from datetime import datetime
    timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
    with open(logfile, "a", encoding="utf-8") as f:
        f.write(f"[{timestamp}] {message}\n")

log_event("Application started")
log_event("Loaded 3 student records")

# ── List Comprehension Basics ──

# Traditional way
scores = [45, 78, 92, 63, 88, 55, 70]
passing = []
for s in scores:
    if s >= 60:
        passing.append(s)

# List comprehension — same result, one line
passing = [s for s in scores if s >= 60]
print("Passing scores:", passing)

# Transform and filter in one step
percentages = [f"{s}%" for s in scores if s >= 60]
print("Percentages:", percentages)

# ── Working with lists of dicts (common in AI/data work) ──

students = [
    {"name": "Arjun", "score": 85, "grade": "B"},
    {"name": "Priya", "score": 92, "grade": "A"},
    {"name": "Ravi",  "score": 55, "grade": "F"},
    {"name": "Meera", "score": 78, "grade": "C"},
]

# Get names of students who passed
passed_names = [s["name"] for s in students if s["score"] >= 60]
print("Passed:", passed_names)

# Normalize all scores to 0-1 range
max_score = max(s["score"] for s in students)
normalized = [{**s, "normalized": round(s["score"] / max_score, 2)} for s in students]
print("Normalized:", normalized)

# ── Dict comprehension ──
name_to_score = {s["name"]: s["score"] for s in students}
print("Name map:", name_to_score)

# ── Sorting ──
by_score = sorted(students, key=lambda s: s["score"], reverse=True)
print("Top scorer:", by_score[0]["name"])

# ── Grouping by grade ──
from collections import defaultdict
by_grade = defaultdict(list)
for s in students:
    by_grade[s["grade"]].append(s["name"])

print("By grade:", dict(by_grade))

import logging
from pathlib import Path

# ── Set up logging (do this once at app startup) ──
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s [%(levelname)s] %(name)s: %(message)s",
    handlers=[
        logging.StreamHandler(),                          # Print to console
        logging.FileHandler("app.log", encoding="utf-8") # Write to file
    ]
)
logger = logging.getLogger("StudyApp")

# ── Error handling patterns ──

def load_config(filepath: str) -> dict:
    """Load config file with proper error handling."""
    import json
    path = Path(filepath)
    
    if not path.exists():
        logger.warning(f"Config file not found: {filepath}. Using defaults.")
        return {"model": "gpt-3.5-turbo", "temperature": 0}
    
    try:
        with open(path, "r") as f:
            config = json.load(f)
        logger.info(f"Config loaded from {filepath}")
        return config
        
    except json.JSONDecodeError as e:
        logger.error(f"Config file is invalid JSON: {e}")
        raise ValueError(f"Cannot parse config: {filepath}") from e
    
    except PermissionError:
        logger.error(f"No permission to read: {filepath}")
        raise

def process_student_score(raw_score: str) -> float:
    """Convert a raw score string to float. Handle bad input."""
    try:
        score = float(raw_score)
        
        if not (0 <= score <= 100):
            raise ValueError(f"Score must be 0-100, got {score}")
        
        logger.debug(f"Processed score: {score}")
        return score
        
    except ValueError as e:
        logger.warning(f"Invalid score '{raw_score}': {e}")
        return 0.0  # Return safe default instead of crashing

# Test error handling
config = load_config("config.json")  # File doesn't exist → returns defaults
print("Config:", config)

for raw in ["85", "99.5", "abc", "150", "-5"]:
    score = process_student_score(raw)
    print(f"  '{raw}' → {score}")

import asyncio
import aiohttp  # pip install aiohttp — for async HTTP requests
from typing import Optional

# ── Async API calls in parallel ──

async def fetch_topic_info(session: aiohttp.ClientSession, topic: str) -> dict:
    """Async HTTP fetch — does not block while waiting for response."""
    url = f"https://en.wikipedia.org/api/rest_v1/page/summary/{topic}"
    
    try:
        async with session.get(url, timeout=aiohttp.ClientTimeout(total=5)) as response:
            if response.status == 200:
                data = await response.json()
                return {"topic": topic, "summary": data.get("extract", "")[:100]}
            return {"topic": topic, "summary": "Not found"}
    except Exception as e:
        return {"topic": topic, "error": str(e)}

async def fetch_all_topics(topics: list) -> list:
    """Fetch all topics in parallel — much faster than sequential."""
    async with aiohttp.ClientSession() as session:
        tasks = [fetch_topic_info(session, topic) for topic in topics]
        results = await asyncio.gather(*tasks, return_exceptions=True)
    return results

# ── Async with LangChain ──

async def run_parallel_chains(queries: list) -> list:
    """Run multiple LangChain chains concurrently."""
    from langchain.chat_models import ChatOpenAI
    from langchain.schema import HumanMessage
    
    llm = ChatOpenAI(model="gpt-3.5-turbo", temperature=0)
    
    async def single_query(q: str) -> str:
        response = await llm.apredict(q)
        return response
    
    tasks = [single_query(q) for q in queries]
    return await asyncio.gather(*tasks)

# Run async functions
topics = ["Python", "Machine_learning", "Artificial_intelligence"]
results = asyncio.run(fetch_all_topics(topics))
for r in results:
    print(f"{r.get('topic')}: {r.get('summary', r.get('error', ''))[:60]}")

# study_utils.py — the module we are testing

def calculate_grade(score: float) -> str:
    """Convert a numeric score to a letter grade."""
    if score < 0 or score > 100:
        raise ValueError(f"Score must be 0-100, got {score}")
    if score >= 90: return "A"
    if score >= 80: return "B"
    if score >= 70: return "C"
    if score >= 60: return "D"
    return "F"

def summarize_class(scores: list) -> dict:
    """Calculate class statistics."""
    if not scores:
        return {"count": 0, "average": 0, "highest": 0, "passing_rate": 0}
    
    passing = [s for s in scores if s >= 60]
    return {
        "count": len(scores),
        "average": round(sum(scores) / len(scores), 1),
        "highest": max(scores),
        "passing_rate": round(len(passing) / len(scores) * 100, 1)
    }

# ---

# test_study_utils.py — the test file (run with: pytest test_study_utils.py -v)
import pytest
from unittest.mock import patch, MagicMock

# Import the functions to test
# from study_utils import calculate_grade, summarize_class

class TestCalculateGrade:
    def test_a_grade(self):
        assert calculate_grade(95) == "A"
        assert calculate_grade(90) == "A"

    def test_b_grade(self):
        assert calculate_grade(85) == "B"
        assert calculate_grade(80) == "B"

    def test_f_grade(self):
        assert calculate_grade(55) == "F"
        assert calculate_grade(0) == "F"

    def test_invalid_score_raises_error(self):
        with pytest.raises(ValueError):
            calculate_grade(150)

        with pytest.raises(ValueError):
            calculate_grade(-5)

class TestSummarizeClass:
    def test_normal_class(self):
        scores = [85, 92, 78, 55, 67]
        result = summarize_class(scores)
        assert result["count"] == 5
        assert result["average"] == 75.4
        assert result["passing_rate"] == 80.0

    def test_empty_class(self):
        result = summarize_class([])
        assert result["count"] == 0
        assert result["average"] == 0

    def test_all_failing(self):
        result = summarize_class([40, 30, 20])
        assert result["passing_rate"] == 0.0

# Mocking an LLM call in tests
class TestAIFunctions:
    @patch("langchain.chat_models.ChatOpenAI")
    def test_ai_response(self, MockLLM):
        # Set up the mock to return a fixed response
        mock_instance = MockLLM.return_value
        mock_instance.predict.return_value = "Photosynthesis converts light to energy."

        # Your function that uses the LLM
        result = mock_instance.predict("What is photosynthesis?")
        
        assert "photosynthesis" in result.lower()
        assert len(result) > 0
        mock_instance.predict.assert_called_once()