Python

Python reference for core syntax, NumPy arrays, Pandas dataframes, Scikit-learn models, and Airflow pipelines, with runnable code examples.

What this covers

Core Python programming plus the data science and workflow-orchestration stack built on top of it: built-in data structures, file I/O, and exception handling; NumPy for numerical arrays; Pandas for data manipulation; Scikit-learn for machine learning; Apache Airflow for pipeline orchestration; and general tooling such as Requests and virtual environments.

When to use it

Manipulating lists, dictionaries, or sets, or handling files and exceptions in a Python script
Doing numerical computing or linear algebra with NumPy arrays
Loading, cleaning, transforming, or exporting tabular data with Pandas
Building, evaluating, or tuning a machine learning model with Scikit-learn
Orchestrating a scheduled data pipeline or ETL workflow with Apache Airflow
Making HTTP requests to an API or setting up an isolated Python environment with venv/pip

Core Python Concepts

Data Structures and Operations

Lists:

# Creating and manipulating lists
my_list = [1, 2, 3, 4, 5]
my_list.append(6)          # Add element
my_list.insert(0, 0)       # Insert at position
my_list.remove(3)          # Remove element
my_list.pop()              # Remove last element
my_list.sort()             # Sort in place
sorted_list = sorted(my_list)  # Return sorted copy

Dictionaries:

# Dictionary operations
my_dict = {'name': 'John', 'age': 30, 'city': 'New York'}
my_dict['email'] = 'john@example.com'  # Add key-value
del my_dict['city']       # Remove key
my_dict.update({'age': 31, 'job': 'Developer'})  # Update multiple

# Dictionary methods
keys = my_dict.keys()
values = my_dict.values()
items = my_dict.items()

Sets:

# Set operations
set1 = {1, 2, 3, 4}
set2 = {3, 4, 5, 6}

union = set1 | set2        # Union
intersection = set1 & set2 # Intersection
difference = set1 - set2   # Difference
symmetric_diff = set1 ^ set2  # Symmetric difference

File Operations

Reading and Writing Files:

# Reading files
with open('file.txt', 'r') as f:
    content = f.read()          # Read entire file
    lines = f.readlines()       # Read all lines
    line = f.readline()         # Read single line

# Writing files
with open('output.txt', 'w') as f:
    f.write('Hello World\n')
    f.writelines(['Line 1\n', 'Line 2\n'])

# Binary files
with open('image.jpg', 'rb') as f:
    data = f.read()

Error Handling

Exception Handling:

try:
    # Code that might raise an exception
    result = 10 / 0
except ZeroDivisionError as e:
    print(f"Division by zero: {e}")
except Exception as e:
    print(f"General error: {e}")
else:
    print("No exceptions occurred")
finally:
    print("Always executed")

Custom Exceptions:

class CustomError(Exception):
    pass

def validate_age(age):
    if age < 0:
        raise CustomError("Age cannot be negative")
    if age > 150:
        raise CustomError("Age cannot be greater than 150")

NumPy: Numerical Computing

NumPy provides efficient arrays and mathematical operations for scientific computing.

Array Creation and Operations

Basic Arrays:

import numpy as np

# 1D array (vector)
arr1 = np.array([1, 2, 3, 4])
print(arr1)  # [1 2 3 4]

# 2D array (matrix)
arr2 = np.array([[1, 2, 3], [4, 5, 6]])
print(arr2)
# [[1 2 3]
#  [4 5 6]]

# Special arrays
zeros = np.zeros((2, 3))      # 2x3 matrix of zeros
ones = np.ones((3, 3))        # 3x3 matrix of ones
eye = np.eye(3)               # 3x3 identity matrix
random = np.random.rand(2, 3) # 2x3 random matrix (0-1)
normal = np.random.randn(2, 3) # 2x3 normal distribution

# Sequences
arange = np.arange(0, 10, 2)  # [0, 2, 4, 6, 8]
linspace = np.linspace(0, 1, 5)  # 5 values between 0 and 1

Array Properties:

arr = np.array([[1, 2, 3], [4, 5, 6]])

print(arr.shape)  # (2, 3) - dimensions
print(arr.ndim)   # 2 - number of dimensions
print(arr.size)   # 6 - total elements
print(arr.dtype)  # int64 - data type

Indexing and Slicing:

arr = np.array([[1, 2, 3, 4], [5, 6, 7, 8], [9, 10, 11, 12]])

print(arr[0])      # [1 2 3 4] - first row
print(arr[:, 0])   # [1 5 9] - first column
print(arr[1, :])   # [5 6 7 8] - second row
print(arr[0:2, 1:3])  # [[2 3], [6 7]] - submatrix
print(arr[::2])    # Every other row

Mathematical Operations:

arr = np.array([1, 2, 3, 4])

# Element-wise operations
print(arr * 2)     # [2 4 6 8]
print(arr + arr)   # [2 4 6 8]
print(np.sqrt(arr)) # [1. 1.414 1.732 2.]
print(np.exp(arr))  # [2.718 7.389 20.085 54.598]
print(np.log(arr))  # [0. 0.693 1.099 1.386]

# Statistics
print(arr.mean())  # 2.5
print(arr.sum())   # 10
print(arr.max())   # 4
print(arr.min())   # 1
print(arr.std())   # 1.118
print(arr.var())   # 1.25

Array Manipulation:

arr = np.array([[1, 2, 3], [4, 5, 6]])

# Reshaping
reshaped = arr.reshape(3, 2)  # Change to 3x2
flattened = arr.flatten()      # 1D array
transposed = arr.T             # Transpose

# Concatenation
arr2 = np.array([[7, 8, 9]])
concat_rows = np.concatenate([arr, arr2], axis=0)  # Add row
concat_cols = np.concatenate([arr, arr.T], axis=1) # Add column

# Stacking
stacked = np.stack([arr, arr], axis=0)  # New dimension

Linear Algebra:

A = np.array([[1, 2], [3, 4]])
B = np.array([[5, 6], [7, 8]])
v = np.array([1, 2])

# Matrix operations
dot_product = A @ B          # Matrix multiplication
dot_product = np.dot(A, B)   # Alternative
vector_norm = np.linalg.norm(v)  # Vector norm
matrix_inv = np.linalg.inv(A)    # Matrix inverse
determinant = np.linalg.det(A)   # Determinant

# Eigenvalues and eigenvectors
eigenvals, eigenvecs = np.linalg.eig(A)

# Singular value decomposition
U, s, Vt = np.linalg.svd(A)

Pandas: Data Manipulation

Pandas provides powerful data structures for data analysis and manipulation.

DataFrame and Series

Creating DataFrames:

import pandas as pd

# From dictionary
data = {
    'name': ['Alice', 'Bob', 'Charlie'],
    'age': [25, 30, 35],
    'city': ['NYC', 'LA', 'Chicago']
}
df = pd.DataFrame(data)

# From list of dictionaries
data_list = [
    {'name': 'Alice', 'age': 25, 'city': 'NYC'},
    {'name': 'Bob', 'age': 30, 'city': 'LA'},
    {'name': 'Charlie', 'age': 35, 'city': 'Chicago'}
]
df = pd.DataFrame(data_list)

# From CSV
df = pd.read_csv('data.csv')
df = pd.read_excel('data.xlsx')

DataFrame Operations:

# Basic information
print(df.head())      # First 5 rows
print(df.tail())      # Last 5 rows
print(df.info())      # Data types and non-null counts
print(df.describe())  # Statistical summary
print(df.shape)       # Dimensions
print(df.columns)     # Column names
print(df.index)       # Index

Data Selection:

# Column selection
names = df['name']           # Single column (Series)
subset = df[['name', 'age']] # Multiple columns (DataFrame)

# Row selection
first_row = df.loc[0]        # By label
second_row = df.iloc[1]      # By position

# Conditional selection
adults = df[df['age'] > 25]
nyc_people = df[df['city'] == 'NYC']

# Multiple conditions
young_nyc = df[(df['age'] < 30) & (df['city'] == 'NYC')]

Data Modification:

# Adding columns
df['age_in_10_years'] = df['age'] + 10
df['full_name'] = df['first_name'] + ' ' + df['last_name']

# Modifying values
df.loc[0, 'age'] = 26
df['city'] = df['city'].str.upper()

# Renaming columns
df = df.rename(columns={'name': 'full_name', 'city': 'location'})

Data Cleaning:

# Handling missing values
df.dropna()                    # Remove rows with NaN
df.fillna(0)                   # Fill NaN with 0
df.fillna(df.mean())           # Fill with column mean

# Removing duplicates
df.drop_duplicates()

# Data type conversion
df['age'] = df['age'].astype(int)
df['date'] = pd.to_datetime(df['date'])

Grouping and Aggregation:

# Group by single column
grouped = df.groupby('city')
city_counts = grouped.size()
city_avg_age = grouped['age'].mean()

# Group by multiple columns
grouped_multi = df.groupby(['city', 'department'])
stats = grouped_multi['salary'].agg(['mean', 'min', 'max', 'count'])

# Custom aggregation
def custom_agg(x):
    return x.max() - x.min()

result = df.groupby('city')['age'].agg(custom_agg)

Merging and Joining:

# Creating sample DataFrames
df1 = pd.DataFrame({
    'id': [1, 2, 3],
    'name': ['Alice', 'Bob', 'Charlie']
})

df2 = pd.DataFrame({
    'id': [1, 2, 4],
    'salary': [50000, 60000, 70000]
})

# Inner join (default)
inner_join = pd.merge(df1, df2, on='id', how='inner')

# Left join
left_join = pd.merge(df1, df2, on='id', how='left')

# Outer join
outer_join = pd.merge(df1, df2, on='id', how='outer')

# Concatenation
df3 = pd.DataFrame({
    'id': [5, 6],
    'name': ['David', 'Eve']
})

concatenated = pd.concat([df1, df3], ignore_index=True)

Data Export:

# To CSV
df.to_csv('output.csv', index=False)

# To Excel
df.to_excel('output.xlsx', index=False)

# To JSON
df.to_json('output.json', orient='records')

# To SQL
from sqlalchemy import create_engine
engine = create_engine('sqlite:///data.db')
df.to_sql('table_name', engine, index=False, if_exists='replace')

Scikit-learn: Machine Learning

Scikit-learn provides tools for machine learning and data analysis.

Data Preprocessing

Train-Test Split:

from sklearn.model_selection import train_test_split

X_train, X_test, y_train, y_test = train_test_split(
    X, y, test_size=0.2, random_state=42, stratify=y
)

Feature Scaling:

from sklearn.preprocessing import StandardScaler, MinMaxScaler

# Standardization (mean=0, std=1)
scaler = StandardScaler()
X_train_scaled = scaler.fit_transform(X_train)
X_test_scaled = scaler.transform(X_test)

# Normalization (0-1 range)
minmax_scaler = MinMaxScaler()
X_train_normalized = minmax_scaler.fit_transform(X_train)
X_test_normalized = minmax_scaler.transform(X_test)

Encoding Categorical Variables:

from sklearn.preprocessing import LabelEncoder, OneHotEncoder

# Label encoding
label_encoder = LabelEncoder()
y_encoded = label_encoder.fit_transform(y_categorical)

# One-hot encoding
onehot_encoder = OneHotEncoder(sparse_output=False)  # `sparse` was renamed `sparse_output` in scikit-learn 1.2
X_encoded = onehot_encoder.fit_transform(X_categorical)

Model Training and Evaluation

Classification Example:

from sklearn.datasets import load_iris
from sklearn.model_selection import train_test_split
from sklearn.preprocessing import StandardScaler
from sklearn.ensemble import RandomForestClassifier
from sklearn.metrics import accuracy_score, classification_report, confusion_matrix

# Load data
iris = load_iris()
X, y = iris.data, iris.target

# Split data
X_train, X_test, y_train, y_test = train_test_split(
    X, y, test_size=0.2, random_state=42
)

# Scale features
scaler = StandardScaler()
X_train_scaled = scaler.fit_transform(X_train)
X_test_scaled = scaler.transform(X_test)

# Train model
model = RandomForestClassifier(n_estimators=100, random_state=42)
model.fit(X_train_scaled, y_train)

# Make predictions
y_pred = model.predict(X_test_scaled)

# Evaluate model
print("Accuracy:", accuracy_score(y_test, y_pred))
print("Classification Report:")
print(classification_report(y_test, y_pred))
print("Confusion Matrix:")
print(confusion_matrix(y_test, y_pred))

Regression Example:

from sklearn.datasets import fetch_california_housing
from sklearn.linear_model import LinearRegression
from sklearn.metrics import mean_squared_error, r2_score

# Load data (load_boston was removed in scikit-learn 1.2 over ethical concerns with the dataset)
housing = fetch_california_housing()
X, y = housing.data, housing.target

# Split data
X_train, X_test, y_train, y_test = train_test_split(
    X, y, test_size=0.2, random_state=42
)

# Train model
regressor = LinearRegression()
regressor.fit(X_train, y_train)

# Make predictions
y_pred = regressor.predict(X_test)

# Evaluate model
print("Mean Squared Error:", mean_squared_error(y_test, y_pred))
print("R² Score:", r2_score(y_test, y_pred))

Cross-Validation

K-Fold Cross-Validation:

from sklearn.model_selection import cross_val_score, KFold

# K-fold cross-validation
kf = KFold(n_splits=5, shuffle=True, random_state=42)
scores = cross_val_score(model, X, y, cv=kf, scoring='accuracy')

print("Cross-validation scores:", scores)
print("Mean accuracy:", scores.mean())
print("Standard deviation:", scores.std())

Grid Search for Hyperparameter Tuning:

from sklearn.model_selection import GridSearchCV

# Define parameter grid
param_grid = {
    'n_estimators': [50, 100, 200],
    'max_depth': [None, 10, 20, 30],
    'min_samples_split': [2, 5, 10]
}

# Grid search
grid_search = GridSearchCV(
    RandomForestClassifier(random_state=42),
    param_grid,
    cv=5,
    scoring='accuracy',
    n_jobs=-1
)

grid_search.fit(X_train, y_train)

print("Best parameters:", grid_search.best_params_)
print("Best score:", grid_search.best_score_)

Clustering

K-Means Clustering:

from sklearn.cluster import KMeans
from sklearn.datasets import make_blobs

# Generate sample data
X, _ = make_blobs(n_samples=300, centers=4, cluster_std=0.60, random_state=0)

# K-means clustering
kmeans = KMeans(n_clusters=4, random_state=42)
clusters = kmeans.fit_predict(X)

# Get cluster centers
centers = kmeans.cluster_centers_

Dimensionality Reduction

Principal Component Analysis (PCA):

from sklearn.decomposition import PCA

# PCA for dimensionality reduction
pca = PCA(n_components=2)
X_pca = pca.fit_transform(X_scaled)

# Explained variance ratio
print("Explained variance ratio:", pca.explained_variance_ratio_)
print("Total explained variance:", sum(pca.explained_variance_ratio_))

Apache Airflow: Workflow Orchestration

Apache Airflow is a platform for programmatically authoring, scheduling, and monitoring workflows. It’s widely used for data pipelines, ETL processes, and task automation.

Installation

# Install Airflow
pip install apache-airflow

# Initialize database (SQLite for development)
airflow db init

# Create admin user
airflow users create \
    --username admin \
    --firstname Admin \
    --lastname User \
    --role Admin \
    --email admin@example.com

# Start web server
airflow webserver --port 8080

# Start scheduler (in another terminal)
airflow scheduler

Core Concepts

DAG (Directed Acyclic Graph): A collection of tasks with dependencies Task: A single unit of work Operator: A template for a task (defines what to do) Sensor: A special type of operator that waits for a condition

Basic DAG Structure

from datetime import datetime, timedelta
from airflow import DAG
from airflow.operators.bash import BashOperator
from airflow.operators.python import PythonOperator

# Default arguments for the DAG
default_args = {
    'owner': 'data_team',
    'depends_on_past': False,
    'start_date': datetime(2023, 1, 1),
    'email_on_failure': False,
    'email_on_retry': False,
    'retries': 1,
    'retry_delay': timedelta(minutes=5),
}

# Create DAG
dag = DAG(
    'example_dag',
    default_args=default_args,
    description='A simple example DAG',
    schedule_interval=timedelta(days=1),  # Run daily
    catchup=False,
)

# Define tasks
def print_hello():
    print("Hello from Airflow!")

task1 = PythonOperator(
    task_id='print_hello',
    python_callable=print_hello,
    dag=dag,
)

task2 = BashOperator(
    task_id='print_date',
    bash_command='date',
    dag=dag,
)

# Set task dependencies
task1 >> task2  # task1 runs before task2

Common Operators

BashOperator: Execute bash commands

bash_task = BashOperator(
    task_id='run_script',
    bash_command='python /path/to/script.py',
    dag=dag,
)

PythonOperator: Execute Python functions

def process_data():
    import pandas as pd
    # Data processing logic
    df = pd.read_csv('input.csv')
    # Process data...
    df.to_csv('output.csv')

python_task = PythonOperator(
    task_id='process_data',
    python_callable=process_data,
    dag=dag,
)

EmailOperator: Send emails

from airflow.operators.email import EmailOperator

email_task = EmailOperator(
    task_id='send_email',
    to='recipient@example.com',
    subject='Pipeline Complete',
    html_content='<p>The data pipeline has completed successfully.</p>',
    dag=dag,
)

HttpSensor: Wait for HTTP endpoint

from airflow.providers.http.sensors.http import HttpSensor

http_sensor = HttpSensor(
    task_id='wait_for_api',
    http_conn_id='http_default',
    endpoint='api/v1/status',
    request_params={},
    response_check=lambda response: response.json()['status'] == 'ready',
    poke_interval=30,  # Check every 30 seconds
    timeout=600,       # Timeout after 10 minutes
    dag=dag,
)

Data Pipeline Example

from airflow import DAG
from airflow.operators.python import PythonOperator
from airflow.providers.postgres.operators.postgres import PostgresOperator
from airflow.providers.postgres.hooks.postgres import PostgresHook
from datetime import datetime, timedelta
import pandas as pd

default_args = {
    'owner': 'data_engineer',
    'depends_on_past': False,
    'start_date': datetime(2023, 1, 1),
    'retries': 1,
    'retry_delay': timedelta(minutes=5),
}

dag = DAG(
    'etl_pipeline',
    default_args=default_args,
    description='ETL pipeline for customer data',
    schedule_interval='@daily',
    catchup=False,
)

def extract_data():
    """Extract data from source"""
    # Simulate data extraction
    data = {
        'customer_id': [1, 2, 3, 4, 5],
        'name': ['Alice', 'Bob', 'Charlie', 'David', 'Eve'],
        'email': ['alice@example.com', 'bob@example.com',
                 'charlie@example.com', 'david@example.com', 'eve@example.com']
    }
    df = pd.DataFrame(data)
    df.to_csv('/tmp/customers_raw.csv', index=False)
    return 'Data extracted successfully'

def transform_data():
    """Transform and clean data"""
    df = pd.read_csv('/tmp/customers_raw.csv')

    # Data cleaning
    df['name'] = df['name'].str.title()
    df['email'] = df['email'].str.lower()

    # Add derived columns
    df['domain'] = df['email'].str.split('@').str[1]
    df['signup_date'] = datetime.now().date()

    df.to_csv('/tmp/customers_clean.csv', index=False)
    return 'Data transformed successfully'

def load_data():
    """Load data to database"""
    df = pd.read_csv('/tmp/customers_clean.csv')

    # Connect to PostgreSQL
    hook = PostgresHook(postgres_conn_id='postgres_default')

    # Create table if not exists
    create_table_sql = """
    CREATE TABLE IF NOT EXISTS customers (
        customer_id INTEGER PRIMARY KEY,
        name VARCHAR(100),
        email VARCHAR(100),
        domain VARCHAR(100),
        signup_date DATE
    );
    """
    hook.run(create_table_sql)

    # Insert data
    for _, row in df.iterrows():
        insert_sql = f"""
        INSERT INTO customers (customer_id, name, email, domain, signup_date)
        VALUES ({row['customer_id']}, '{row['name']}', '{row['email']}',
                '{row['domain']}', '{row['signup_date']}')
        ON CONFLICT (customer_id) DO UPDATE SET
            name = EXCLUDED.name,
            email = EXCLUDED.email,
            domain = EXCLUDED.domain;
        """
        hook.run(insert_sql)

    return 'Data loaded successfully'

# Define tasks
extract_task = PythonOperator(
    task_id='extract',
    python_callable=extract_data,
    dag=dag,
)

transform_task = PythonOperator(
    task_id='transform',
    python_callable=transform_data,
    dag=dag,
)

load_task = PythonOperator(
    task_id='load',
    python_callable=load_data,
    dag=dag,
)

# Data quality check
data_quality_check = PostgresOperator(
    task_id='data_quality_check',
    postgres_conn_id='postgres_default',
    sql="""
    SELECT COUNT(*) as record_count FROM customers
    WHERE customer_id IS NOT NULL;
    """,
    dag=dag,
)

# Set dependencies
extract_task >> transform_task >> load_task >> data_quality_check

Scheduling and Triggers

Schedule Intervals:

# Common schedule intervals
schedule_interval='@daily'        # Every day at midnight
schedule_interval='@hourly'       # Every hour
schedule_interval='@weekly'       # Every week
schedule_interval='0 9 * * *'     # Every day at 9 AM
schedule_interval='*/30 * * * *'  # Every 30 minutes
schedule_interval=None            # Manual only

Manual Trigger:

# Trigger DAG manually
airflow dags trigger example_dag

# Trigger with configuration
airflow dags trigger example_dag \
    --conf '{"key": "value"}'

Variables and Connections

Airflow Variables:

from airflow.models import Variable

# Set variable via UI or CLI
# airflow variables set my_var "my_value"

# Get variable in code
my_var = Variable.get("my_var")

# Set variable programmatically
Variable.set("processed_count", 100)

Connections:

from airflow.hooks.base import BaseHook

# Get connection
conn = BaseHook.get_connection('postgres_default')
host = conn.host
user = conn.login
password = conn.password
database = conn.schema

Best Practices

Idempotent Tasks: Ensure tasks can be run multiple times safely
Atomic Operations: Use transactions for database operations
Error Handling: Implement proper error handling and retries
Logging: Use Airflow’s logging for debugging
Modular DAGs: Break complex workflows into smaller, reusable DAGs
Testing: Test DAGs locally before deployment
Documentation: Document DAGs and tasks with docstrings
Resource Management: Set appropriate resource limits
Monitoring: Use Airflow’s web UI for monitoring
Version Control: Keep DAGs in version control

Testing DAGs

from airflow.utils.test import get_test_dag_bag

def test_dag():
    """Test DAG structure and dependencies"""
    dag_bag = get_test_dag_bag()
    dag = dag_bag.get_dag('example_dag')

    assert dag is not None
    assert len(dag.tasks) == 3

    # Test task dependencies
    task1 = dag.get_task('task1')
    task2 = dag.get_task('task2')
    assert task2 in task1.downstream_list

if __name__ == '__main__':
    test_dag()

Deployment Considerations

Production Setup:

Use external database (PostgreSQL/MySQL) instead of SQLite
Configure proper logging and monitoring
Set up proper authentication and authorization
Use connection pools for database connections
Implement proper backup and recovery procedures

Scaling:

Use CeleryExecutor for distributed task execution
Configure worker nodes for parallel processing
Monitor resource usage and adjust accordingly

Additional Libraries

Requests: HTTP Library

import requests

# GET request
response = requests.get('https://api.example.com/data')
print(response.status_code)
print(response.json())

# POST request
data = {'key': 'value'}
response = requests.post('https://api.example.com/submit', json=data)

# With authentication
response = requests.get('https://api.example.com/protected',
                       auth=('username', 'password'))

# Headers and timeout
headers = {'Authorization': 'Bearer token'}
response = requests.get('https://api.example.com/data',
                       headers=headers, timeout=5)

Virtual Environments

Using venv:

# Create virtual environment
python -m venv myenv

# Activate (Linux/Mac)
source myenv/bin/activate

# Activate (Windows)
myenv\Scripts\activate

# Install packages
pip install package-name

# Deactivate
deactivate

# Export requirements
pip freeze > requirements.txt

# Install from requirements
pip install -r requirements.txt

Best Practices

Code Style: Follow PEP 8 guidelines
Type Hints: Use type annotations for better code documentation
Error Handling: Implement proper exception handling
Testing: Write unit tests with pytest or unittest
Documentation: Use docstrings and comments
Virtual Environments: Always use virtual environments
Dependencies: Keep requirements.txt updated
Performance: Use appropriate data structures and algorithms
Security: Validate inputs and handle sensitive data carefully
Version Control: Use Git for all projects

← Back to Homepage