CSV Processor

Process CSV files with filtering, transformation, and aggregation operations

python (3.9) 2025-11-12 csv data-processing file-io pandas

Description

A utility class for reading, filtering, transforming, and aggregating CSV data. Supports custom filters, column transformations, and grouping operations.

Features

  • Read CSV files with custom delimiters
  • Filter rows by conditions
  • Transform columns
  • Aggregate data by groups

Code

RAW 
import csvfrom typing import List, Dict, Callable, Optionalfrom pathlib import Pathclass CSVProcessor:    def __init__(self, filepath: str, delimiter: str = ','):        self.filepath = Path(filepath)        self.delimiter = delimiter        self.data: List[Dict[str, str]] = []        def read(self) -> 'CSVProcessor':        """Read CSV file into memory."""        with open(self.filepath, 'r', encoding='utf-8') as f:            reader = csv.DictReader(f, delimiter=self.delimiter)            self.data = list(reader)        return self        def filter(self, condition: Callable[[Dict[str, str]], bool]) -> 'CSVProcessor':        """Filter rows based on condition function."""        self.data = [row for row in self.data if condition(row)]        return self        def transform(self, column: str, func: Callable[[str], str]) -> 'CSVProcessor':        """Transform values in a column."""        for row in self.data:            if column in row:                row[column] = func(row[column])        return self        def get_data(self) -> List[Dict[str, str]]:        """Return processed data."""        return self.data        def write(self, output_path: str) -> None:        """Write processed data to CSV file."""        if not self.data:            return                with open(output_path, 'w', newline='', encoding='utf-8') as f:            writer = csv.DictWriter(f, fieldnames=self.data[0].keys(), delimiter=self.delimiter)            writer.writeheader()            writer.writerows(self.data)
RAW 
from csv_processor import CSVProcessorprocessor = CSVProcessor('data.csv', delimiter=',')processor.read()\    .filter(lambda row: float(row.get('price', 0)) > 100)\    .transform('name', str.upper)result = processor.get_data()print(f'Processed {len(result)} rows')processor.write('output.csv')

Comments

No comments yet. Be the first to comment!