Development Workflows

ls /root/.pyenv/versions

# Using Python 3.11.13 (default)
/root/.pyenv/versions/3.11.13/bin/python -m venv my_project_env

# Using specific Python version
/root/.pyenv/versions/3.10.18/bin/python -m venv legacy_project_env

# Activate environment
source my_project_env/bin/activate

# Verify active environment (should show your env path)
which python

# Deactivate when finished
deactivate

# Core data processing
pandas==2.0.3
numpy==1.24.3

# API and web requests  
requests==2.31.0
urllib3==2.0.4

# Visualization
matplotlib==3.7.2
seaborn==0.12.2

# Development tools
jupyter==1.0.0
pytest==7.4.0

# Activate environment first
source my_project_env/bin/activate

# Install from requirements file
pip install -r requirements.txt

# Install additional packages and update requirements
pip install scikit-learn==1.3.0
pip freeze > requirements.txt

import pandas as pd
import numpy as np
from sqlalchemy import create_engine

def process_customer_data(connection_string):
    """Process customer data from warehouse"""
    engine = create_engine(connection_string)
    
    # Load data
    df = pd.read_sql("SELECT * FROM customers", engine)
    
    # Data transformations
    df['full_name'] = df['first_name'] + ' ' + df['last_name']
    df['customer_tier'] = pd.cut(df['total_spent'], 
                                bins=[0, 100, 500, 1000, float('inf')],
                                labels=['Bronze', 'Silver', 'Gold', 'Platinum'])
    
    # Save processed data
    df.to_sql('processed_customers', engine, if_exists='replace', index=False)
    
    return df

if __name__ == "__main__":
    # Your connection string here
    conn_str = "your_connection_string"
    result = process_customer_data(conn_str)
    print(f"Processed {len(result)} customers")

import requests
import json
from datetime import datetime

class DataAPI:
    def __init__(self, base_url, api_key):
        self.base_url = base_url
        self.headers = {'Authorization': f'Bearer {api_key}'}
    
    def fetch_data(self, endpoint, params=None):
        """Fetch data from API endpoint"""
        response = requests.get(
            f"{self.base_url}/{endpoint}",
            headers=self.headers,
            params=params
        )
        response.raise_for_status()
        return response.json()
    
    def process_and_save(self, endpoint, db_connection):
        """Fetch, process, and save data to database"""
        data = self.fetch_data(endpoint)
        
        # Process data
        df = pd.DataFrame(data)
        df['processed_at'] = datetime.now()
        
        # Save to database
        df.to_sql('api_data', db_connection, if_exists='append', index=False)
        
        return df

# List available dbt environments
ls /root/.venv

# Activate specific dbt version
source /root/.venv/dbt-1.8.9/bin/activate

# Verify dbt installation
dbt --version

# Navigate to your dbt project directory
cd /path/to/your/dbt/project

# Run entire project
dbt run

# Run specific models
dbt run --select staging.stg_customers+

# Test your models
dbt test

# Generate documentation
dbt docs generate
dbt docs serve

-- models/staging/stg_customers.sql
SELECT
    customer_id::int AS customer_id,
    LOWER(TRIM(email)) AS email,
    INITCAP(first_name) AS first_name,
    INITCAP(last_name) AS last_name,
    created_at::timestamp AS created_at,
    CASE 
        WHEN status = 'A' THEN 'active'
        WHEN status = 'I' THEN 'inactive'
        ELSE 'unknown'
    END AS status
FROM {{ source('crm', 'customers') }}
WHERE customer_id IS NOT NULL

# models/schema.yml
version: 2

models:
  - name: stg_customers
    description: "Cleaned customer data from CRM system"
    columns:
      - name: customer_id
        description: "Unique customer identifier"
        tests:
          - unique
          - not_null
      - name: email
        description: "Customer email address"
        tests:
          - not_null
          - unique

# Run all models
dbt run

# Run specific models
dbt run --select staging.stg_customers+

# Test models
dbt test

# Generate documentation
dbt docs generate

# Compile models
dbt compile