Dart: Project Design and Development — DataPipeline CLI Tool

> **Output:** Run locally in DartPad or using `dart run`. All examples in the Dart course are based on Dart 3.x / Flutter 3.x, and results may vary slightly depending on the SDK version.

> **Output:** Run locally in DartPad or using `dart run`. All examples in the Dart course are based on Dart 3.x / Flutter 3.x, and results may vary slightly depending on the SDK version.

data_pipeline/
  bin/
    data_pipeline.dart        # CLI entry point
  lib/
    src/
      core/
        models/
          order.dart
          product.dart
          customer.dart
          analysis_result.dart
        utils/
          formatters.dart
          validators.dart
          constants.dart
      data/
        sources/
          data_source.dart     # Sealed class
          csv_source.dart
          json_source.dart
          api_source.dart
        parsers/
          order_parser.dart
          product_parser.dart
      services/
        pipeline.dart
        analyzer.dart
        isolate_pool.dart
        report_generator.dart
    data_pipeline.dart         # Barrel export
  test/
    core/
      models_test.dart
      utils_test.dart
    data/
      parsers_test.dart
    services/
      pipeline_test.dart
      analyzer_test.dart
  pubspec.yaml
```text


```text
> **Output:** Run locally in DartPad or using `dart run`. All examples in the Dart course are based on Dart 3.x / Flutter 3.x, and results may vary slightly depending on the SDK version.

> **Output:** Run locally in DartPad or using `dart run`. All examples in the Dart course are based on Dart 3.x / Flutter 3.x, and results may vary slightly depending on the SDK version.

import 'package:args/args.dart';
import 'package:args/command_runner.dart';

// Sub-command: analyze
class AnalyzeCommand extends Command {
  @override
  final name = 'analyze';
  @override
  final description = 'Analyze order data and generate statistics';

  AnalyzeCommand() {
    argParser
      ..addOption('source', abbr: 's', defaultsTo: 'csv', allowed: ['csv', 'json', 'api'])
      ..addOption('input', abbr: 'i', mandatory: true)
      ..addOption('output', abbr: 'o', defaultsTo: 'stdout')
      ..addFlag('parallel', abbr: 'p', defaultsTo: false)
      ..addOption('isolate-count', defaultsTo: '4');
  }

  @override
  Future``<void>`` run() async {
    final source = argResults!['source'] as String;
    final input = argResults!['input'] as String;
    final output = argResults!['output'] as String;
    final parallel = argResults!['parallel'] as bool;
    final isolateCount = int.parse(argResults!['isolate-count'] as String);

    print('Source: $source | Input: $input | Parallel: $parallel');
  }
}

// Sub-command: export
class ExportCommand extends Command {
  @override
  final name = 'export';
  @override
  final description = 'Export analysis results to file';

  ExportCommand() {
    argParser
      ..addOption('format', defaultsTo: 'json', allowed: ['json', 'csv', 'markdown'])
      ..addOption('input', abbr: 'i', mandatory: true)
      ..addOption('output', abbr: 'o', mandatory: true);
  }

  @override
  Future``<void>`` run() async {
    final format = argResults!['format'] as String;
    final input = argResults!['input'] as String;
    final output = argResults!['output'] as String;

    print('Export: $format | $input -> $output');
  }
}

void main(List``<String>`` args) async {
  final runner = CommandRunner('data_pipeline', 'DataPipeline - E-commerce data analytics CLI')
    ..addCommand(AnalyzeCommand())
    ..addCommand(ExportCommand());

  try {
    await runner.run(args);
  } on UsageException catch (e) {
    print(e);
  }
}

> **Output:** Run locally in DartPad or using `dart run`. All examples in the Dart course are based on Dart 3.x / Flutter 3.x, and results may vary slightly depending on the SDK version.

> **Output:** Run locally in DartPad or using `dart run`. All examples in the Dart course are based on Dart 3.x / Flutter 3.x, and results may vary slightly depending on the SDK version.

sealed class DataSource {
  const DataSource();

  String get displayName;

  Stream``<String>`` readLines();
}

class CsvSource extends DataSource {
  final String path;
  final String delimiter;

  const CsvSource({required this.path, this.delimiter = ','});

  @override
  String get displayName => 'CSV: $path';

  @override
  Stream``<String>`` readLines() => File(path).openRead().transform(utf8.decoder).transform(const LineSplitter());
}

class JsonSource extends DataSource {
  final String path;

  const JsonSource({required this.path});

  @override
  String get displayName => 'JSON: $path';

  @override
  Stream``<String>`` readLines() async* {
    final content = await File(path).readAsString();
    final jsonList = jsonDecode(content) as List;
    for (final item in jsonList) {
      yield jsonEncode(item);
    }
  }
}

class ApiSource extends DataSource {
  final String endpoint;
  final Map<String, String> headers;

  const ApiSource({required this.endpoint, this.headers = const {}});

  @override
  String get displayName => 'API: $endpoint';

  @override
  Stream``<String>`` readLines() async* {
    final client = Client();
    final response = await client.get(Uri.parse(endpoint), headers: headers);
    final jsonList = jsonDecode(response.body) as List;
    for (final item in jsonList) {
      yield jsonEncode(item);
    }
    client.close();
  }
}

> **Output:** Run locally in DartPad or using `dart run`. All examples in the Dart course are based on Dart 3.x / Flutter 3.x, and results may vary slightly depending on the SDK version.

> **Output:** Run locally in DartPad or using `dart run`. All examples in the Dart course are based on Dart 3.x / Flutter 3.x, and results may vary slightly depending on the SDK version.

DataSource createSource(String type, String input) => switch (type) {
  'csv' => CsvSource(path: input),
  'json' => JsonSource(path: input),
  'api' => ApiSource(endpoint: input),
  _ => throw ArgumentError('Unknown source type: $type'),
};

String sourceIcon(DataSource source) => switch (source) {
  CsvSource() => '📄',
  JsonSource() => '📋',
  ApiSource() => '🌐',
};

> **Output:** Run locally in DartPad or using `dart run`. All examples in the Dart course are based on Dart 3.x / Flutter 3.x, and results may vary slightly depending on the SDK version.

> **Output:** Run locally in DartPad or using `dart run`. All examples in the Dart course are based on Dart 3.x / Flutter 3.x, and results may vary slightly depending on the SDK version.

abstract class OrderParser {
  const OrderParser();

  Order parse(String raw);
  List``<Order>`` parseBatch(List``<String>`` raws) => raws.map(parse).toList();
}

class CsvOrderParser extends OrderParser {
  final String delimiter;

  const CsvOrderParser({this.delimiter = ','});

  @override
  Order parse(String raw) {
    final parts = raw.split(delimiter);
    if (parts.length < 4) {
      throw FormatException('Invalid CSV row: $raw');
    }
    return Order(
      id: parts[0].trim(),
      amount: double.parse(parts[1].trim()),
      status: parts[2].trim(),
      category: parts[3].trim(),
      region: parts.length > 4 ? parts[4].trim() : 'US',
    );
  }
}

class JsonOrderParser extends OrderParser {
  const JsonOrderParser();

  @override
  Order parse(String raw) {
    final json = jsonDecode(raw) as Map<String, dynamic>;
    return Order(
      id: json['id'] as String,
      amount: (json['amount'] as num).toDouble(),
      status: json['status'] as String,
      category: json['category'] as String,
      region: (json['region'] as String?) ?? 'US',
    );
  }
}

> **Output:** Run locally in DartPad or using `dart run`. All examples in the Dart course are based on Dart 3.x / Flutter 3.x, and results may vary slightly depending on the SDK version.

> **Output:** Run locally in DartPad or using `dart run`. All examples in the Dart course are based on Dart 3.x / Flutter 3.x, and results may vary slightly depending on the SDK version.

import 'dart:async';
import 'dart:convert';
import 'dart:io';

class Pipeline {
  final DataSource _source;
  final OrderParser _parser;
  final int _chunkSize;

  Pipeline({
    required DataSource source,
    required OrderParser parser,
    int chunkSize = 1000,
  })  : _source = source,
        _parser = parser,
        _chunkSize = chunkSize;

  Future``<AnalysisResult>`` execute() async {
    final orders = ``<Order>``[];
    var processed = 0;
    var skipped = 0;

    await for (final line in _source.readLines()) {
      try {
        final order = _parser.parse(line);
        orders.add(order);
        processed++;
      } on FormatException {
        skipped++;
      }

      if (processed % _chunkSize == 0) {
        stdout.writeln('Progress: $processed orders processed, $skipped skipped');
      }
    }

    stdout.writeln('Total: $processed processed, $skipped skipped');

    final analyzer = OrderAnalyzer();
    return analyzer.analyze(orders);
  }

  Stream``<Order>`` streamOrders() async* {
    await for (final line in _source.readLines()) {
      try {
        yield _parser.parse(line);
      } on FormatException {
        continue;
      }
    }
  }

  Stream``<AnalysisResult>`` streamByCategory() {
    final controller = StreamController``<AnalysisResult>``();

    streamOrders().fold<Map<String, List``<Order>``>>(
      {},
      (groups, order) {
        groups.update(order.category, (v) => v..add(order), ifAbsent: () => [order]);
        return groups;
      },
    ).then((groups) {
      final analyzer = OrderAnalyzer();
      for (final entry in groups.entries) {
        controller.add(analyzer.analyze(entry.value));
      }
      controller.close();
    });

    return controller.stream;
  }
}

> **Output:** Run locally in DartPad or using `dart run`. All examples in the Dart course are based on Dart 3.x / Flutter 3.x, and results may vary slightly depending on the SDK version.

> **Output:** Run locally in DartPad or using `dart run`. All examples in the Dart course are based on Dart 3.x / Flutter 3.x, and results may vary slightly depending on the SDK version.

import 'dart:isolate';

class IsolatePool {
  final int _poolSize;
  final List``<Isolate>`` _isolates = [];
  final List``<SendPort>`` _sendPorts = [];

  IsolatePool({int poolSize = 4}) : _poolSize = poolSize;

  Future``<void>`` initialize() async {
    for (var i = 0; i < _poolSize; i++) {
      final receivePort = ReceivePort();
      final isolate = await Isolate.spawn(
        _isolateEntryPoint,
        receivePort.sendPort,
      );
      final sendPort = await receivePort.first as SendPort;
      _isolates.add(isolate);
      _sendPorts.add(sendPort);
    }
  }

  static void _isolateEntryPoint(SendPort mainSendPort) {
    final receivePort = ReceivePort();
    mainSendPort.send(receivePort.sendPort);

    receivePort.listen((message) {
      if (message is _IsolateTask) {
        final result = _processChunk(message.orders, message.taxRate);
        message.responsePort.send(result);
      }
    });
  }

  static AnalysisResult _processChunk(List<Map<String, dynamic>> rawOrders, double taxRate) {
    final orders = rawOrders.map((o) => Order(
      id: o['id'] as String,
      amount: (o['amount'] as num).toDouble(),
      status: o['status'] as String,
      category: o['category'] as String,
      region: (o['region'] as String?) ?? 'US',
    )).toList();

    final analyzer = OrderAnalyzer(taxRate: taxRate);
    return analyzer.analyze(orders);
  }

  Future<List``<AnalysisResult>``> processInParallel(
    List<Map<String, dynamic>> allOrders,
    double taxRate,
  ) async {
    final chunkSize = (allOrders.length / _poolSize).ceil();
    final results = ``<AnalysisResult>``[];
    final completers = <Completer``<AnalysisResult>``>[];

    for (var i = 0; i < _poolSize; i++) {
      final start = i * chunkSize;
      final end = (start + chunkSize).clamp(0, allOrders.length);
      if (start >= allOrders.length) break;

      final chunk = allOrders.sublist(start, end);
      final completer = Completer``<AnalysisResult>``();
      completers.add(completer);

      _sendPorts[i].send(_IsolateTask(
        orders: chunk,
        taxRate: taxRate,
        responsePort: completer.future as dynamic,
      ));
    }

    for (final completer in completers) {
      results.add(await completer.future);
    }

    return results;
  }

  void dispose() {
    for (final isolate in _isolates) {
      isolate.kill(priority: Isolate.immediate);
    }
  }
}

class _IsolateTask {
  final List<Map<String, dynamic>> orders;
  final double taxRate;
  final SendPort responsePort;

  const _IsolateTask({
    required this.orders,
    required this.taxRate,
    required this.responsePort,
  });
}

> **Output:** Run locally in DartPad or using `dart run`. All examples in the Dart course are based on Dart 3.x / Flutter 3.x, and results may vary slightly depending on the SDK version.

> **Output:** Run locally in DartPad or using `dart run`. All examples in the Dart course are based on Dart 3.x / Flutter 3.x, and results may vary slightly depending on the SDK version.

class ServiceLocator {
  static final _instances = <Type, dynamic>{};

  static void register``<T>``(T instance) {
    _instances[T] = instance;
  }

  static T get``<T>``() {
    final instance = _instances[T];
    if (instance == null) {
      throw StateError('Service not registered: $T');
    }
    return instance as T;
  }

  static void reset() {
    _instances.clear();
  }
}

// Registration at startup
void setupServices({required String sourceType, required String inputPath}) {
  final source = createSource(sourceType, inputPath);

  ServiceLocator.register``<DataSource>``(source);

  final parser = switch (source) {
    CsvSource() => const CsvOrderParser() as OrderParser,
    JsonSource() => const JsonOrderParser(),
    ApiSource() => const JsonOrderParser(),
  };

  ServiceLocator.register``<OrderParser>``(parser);
  ServiceLocator.register``<OrderAnalyzer>``(const OrderAnalyzer());
  ServiceLocator.register``<Pipeline>``(Pipeline(source: source, parser: parser));
}

> **Output:** Run locally in DartPad or using `dart run`. All examples in the Dart course are based on Dart 3.x / Flutter 3.x, and results may vary slightly depending on the SDK version.

// ============================================
// DataPipeline CLI Tool - Complete Implementation
// Bob's e-commerce data analytics tool
// ============================================

import 'dart:async';
import 'dart:convert';
import 'dart:io';
import 'dart:isolate';

// ---- Core Models ----

class Order {
  final String id;
  final double amount;
  final String status;
  final String category;
  final String region;

  const Order({
    required this.id,
    required this.amount,
    required this.status,
    required this.category,
    this.region = 'US',
  });

  double get tax => amount * _taxRate(region);
  double get total => amount + tax;

  static double _taxRate(String region) => switch (region) {
    'US' => 0.08,
    'EU' => 0.20,
    'UK' => 0.15,
    'JP' => 0.10,
    _ => 0.10,
  };

  Map<String, dynamic> toJson() => {
    'id': id,
    'amount': amount,
    'status': status,
    'category': category,
    'region': region,
  };
}

class AnalysisResult {
  final int totalOrders;
  final int completedOrders;
  final double revenue;
  final double tax;
  final double total;
  final Map<String, double> revenueByCategory;

  const AnalysisResult({
    required this.totalOrders,
    required this.completedOrders,
    required this.revenue,
    required this.tax,
    required this.total,
    required this.revenueByCategory,
  });

  double get averageOrderValue => completedOrders > 0 ? revenue / completedOrders : 0;

  @override
  String toString() => '''
=== DataPipeline Analytics ===
Orders:    $completedOrders/$totalOrders completed
Revenue:   \$${revenue.toStringAsFixed(2)} USD
Tax:       \$${tax.toStringAsFixed(2)} USD
Total:     \$${total.toStringAsFixed(2)} USD
Average:   \$${averageOrderValue.toStringAsFixed(2)} USD
Categories: ${revenueByCategory.keys.join(', ')}
''';
}

// ---- Core Services ----

class OrderAnalyzer {
  final double taxRate;

  const OrderAnalyzer({this.taxRate = 0.08});

  AnalysisResult analyze(List``<Order>`` orders) {
    final completed = orders.where((o) => o.status == 'completed').toList();
    final revenue = completed.fold``<double>``(0, (s, o) => s + o.amount);
    final tax = completed.fold``<double>``(0, (s, o) => s + o.tax);
    final total = revenue + tax;

    final byCategory = <String, double>{};
    for (final o in completed) {
      byCategory.update(o.category, (v) => v + o.amount, ifAbsent: () => o.amount);
    }

    return AnalysisResult(
      totalOrders: orders.length,
      completedOrders: completed.length,
      revenue: revenue,
      tax: tax,
      total: total,
      revenueByCategory: byCategory,
    );
  }
}

// ---- Data Source (Sealed) ----

sealed class DataSource {
  const DataSource();
  String get displayName;
  Stream``<String>`` readLines();
}

class CsvSource extends DataSource {
  final String path;
  const CsvSource({required this.path});

  @override
  String get displayName => 'CSV: $path';

  @override
  Stream``<String>`` readLines() =>
    File(path).openRead().transform(utf8.decoder).transform(const LineSplitter());
}

class JsonSource extends DataSource {
  final String path;
  const JsonSource({required this.path});

  @override
  String get displayName => 'JSON: $path';

  @override
  Stream``<String>`` readLines() async* {
    final content = await File(path).readAsString();
    final list = jsonDecode(content) as List;
    for (final item in list) {
      yield jsonEncode(item);
    }
  }
}

// ---- Parser ----

abstract class OrderParser {
  const OrderParser();
  Order parse(String raw);
}

class CsvOrderParser extends OrderParser {
  const CsvOrderParser();

  @override
  Order parse(String raw) {
    final parts = raw.split(',');
    if (parts.length < 4) throw FormatException('Invalid row: $raw');
    return Order(
      id: parts[0].trim(),
      amount: double.parse(parts[1].trim()),
      status: parts[2].trim(),
      category: parts[3].trim(),
    );
  }
}

class JsonOrderParser extends OrderParser {
  const JsonOrderParser();

  @override
  Order parse(String raw) {
    final json = jsonDecode(raw) as Map<String, dynamic>;
    return Order(
      id: json['id'] as String,
      amount: (json['amount'] as num).toDouble(),
      status: json['status'] as String,
      category: json['category'] as String,
    );
  }
}

// ---- Pipeline ----

class Pipeline {
  final DataSource source;
  final OrderParser parser;

  const Pipeline({required this.source, required this.parser});

  Future``<AnalysisResult>`` execute() async {
    final orders = ``<Order>``[];
    var count = 0;

    await for (final line in source.readLines()) {
      try {
        orders.add(parser.parse(line));
        count++;
        if (count % 100000 == 0) {
          stdout.writeln('Progress: ${(count / 1000).toStringAsFixed(0)}K orders');
        }
      } on FormatException {
        continue;
      }
    }

    stdout.writeln('Loaded: ${(count / 1000).toStringAsFixed(0)}K orders from ${source.displayName}');
    return const OrderAnalyzer().analyze(orders);
  }
}

// ---- Demo Data Generator ----

List``<String>`` generateDemoOrders(int count) {
  final categories = ['Electronics', 'Clothing', 'Books', 'Home', 'Sports'];
  final statuses = ['completed', 'completed', 'completed', 'pending', 'cancelled'];
  final regions = ['US', 'EU', 'UK', 'JP'];

  return List.generate(count, (i) {
    final category = categories[i % categories.length];
    final status = statuses[i % statuses.length];
    final region = regions[i % regions.length];
    final amount = (100 + (i * 37) % 5000).toDouble();
    return 'ORD-${(i + 1).toString().padLeft(5, '0')},$amount,$status,$category,$region';
  });
}

// ---- Main ----

void main() async {
  print('=== DataPipeline CLI v1.0 ===\n');

  final demoData = generateDemoOrders(1000);
  final tempFile = File('${Directory.systemTemp.path}/demo_orders.csv');
  await tempFile.writeAsString(demoData.join('\n'));

  final source = CsvSource(path: tempFile.path);
  final parser = const CsvOrderParser();
  final pipeline = Pipeline(source: source, parser: parser);

  print('Source: ${source.displayName}');
  print('Processing...\n');

  final result = await pipeline.execute();
  print(result);

  final sorted = result.revenueByCategory.entries.toList()
    ..sort((a, b) => b.value.compareTo(a.value));

  print('Revenue by Category:');
  for (final entry in sorted) {
    print('  ${entry.key.padRight(12)}: \$${entry.value.toStringAsFixed(2)} USD');
  }

  await tempFile.delete();
  print('\nDone. Temporary file cleaned up.');
}

> **Output:** Run locally in DartPad or using `dart run`. All examples in the Dart course are based on Dart 3.x / Flutter 3.x, and results may vary slightly depending on the SDK version.