koruspy 0.9.0

Uma biblioteca inspirada em Rust e Kotlin para lidar com Option, println colorido e utilitários.

Koruspy

Segurança de tipos, performance extrema e código elegante para Python.

Koruspy é uma biblioteca de utilitários funcionais que traz padrões battle-tested de Rust e Scala para o ecossistema Python. Ela elimina erros de NoneType e exceções não tratadas através de Monads (Option e Result), enquanto oferece performance de nível C via Cython e processamento lazy de dados com consumo de memória O(1).

---

🎯 Por que Koruspy?

A biblioteca resolve três problemas críticos do desenvolvimento Python:

1. Robustez: Elimine NoneType errors e exceções silenciosas

# ❌ Python tradicional: código defensivo e verboso
def get_user_email(user_id):
    user = db.find_user(user_id)
    if user is not None:
        if hasattr(user, 'profile'):
            if user.profile is not None:
                return user.profile.email.lower()
    return "[e-mail]"

# ✅ Koruspy: seguro, conciso e legível
def get_user_email(user_id):
    return (option_of(db.find_user(user_id))
            .and_then(lambda u: option_of(u.profile))
            .getattr("email")
            .map(str.lower)
            .unwrap_or("[e-mail]"))

2. Performance: Processamento lazy + motor Cython

# ❌ Python tradicional: estoura memória e IO lento
logs = [i for i in range(1_000_000)]  # Lista inteira na RAM
filtered = [log * 2 for log in logs if log % 10 == 0]
for item in filtered:
    print(item)  # print() nativo é lento em alto volume

# ✅ Koruspy: RAM estável O(1) e impressão ultra-rápida
from koruspy import LazyList, println

(LazyList(lambda: range(1_000_000))
 .lazy_filter(lambda x: x % 10 == 0)
 .lazy_map(lambda x: x * 2)
 .for_each(println))  # Motor Cython com printf direto

Benchmark: Processar e imprimir 100k itens em ~0.87s (vs ~3.2s com print() nativo).

3. Ergonomia: API fluida que parece linguagem natural

# ✅ Encadeamento assíncrono sem await intermediário
async def get_active_profile(user_id):
    return await (async_option(fetch_user(user_id))
                  .map_async(lambda u: u.profile)
                  .filter_async(lambda p: p.is_active)
                  .map_async(lambda p: p.to_dict())
                  .unwrap_or_async({"status": "inactive"}))

---

📦 Instalação

pip install koruspy

---

🔥 Core: Option e Result Pattern

Option: Presença ou Ausência de Valor

Substitui checagens manuais de None por um container semântico.

from koruspy import option_of, Some, nothing

# Criação
user = option_of(db.get_user(123))  # Some(User) ou nothing

# Transformação segura
email = (user
         .map(lambda u: u.email)
         .map(str.upper)
         .unwrap_or("[e-mail]"))

# Filtragem
active_user = user.filter(lambda u: u.is_active)

# Encadeamento condicional
profile_pic = user.and_then(lambda u: option_of(u.profile_url))

Métodos principais:

• map(fn): Transforma o valor interno se existir
• and_then(fn): Encadeia operações que retornam Option
• filter(predicate): Mantém o valor apenas se o predicado for verdadeiro
• unwrap_or(default): Extrai o valor ou retorna default
• getattr(name): Acesso seguro a atributos

Debug visual: Some aparece em verde, nothing em vermelho no terminal.

Result: Sucesso ou Falha

Transforma exceções em valores tratáveis.

from koruspy import result_of, Okay, Err

# Captura automática de exceções
def divide(a, b):
    return result_of(lambda: a / b)

result = divide(10, 2)  # Okay(5.0)
error = divide(10, 0)   # Err(ZeroDivisionError)

# Pipeline de tratamento
processed = (divide(100, 5)
             .map(lambda x: x * 2)
             .flat_map(lambda x: divide(x, 4))
             .unwrap_or(0))  # 10.0

# Matching explícito
match result:
    case Okay(value):
        print(f"Sucesso: {value}")
    case Err(error):
        print(f"Erro: {error}")

Métodos principais:

• map(fn): Transforma o valor em caso de sucesso
• flat_map(fn): Encadeia operações que retornam Result
• unwrap_or(default): Extrai o valor ou retorna default em caso de erro
• is_ok() / is_err(): Checagem de estado

Debug visual: Okay em verde, Err em vermelho com stack trace formatado.

---

⚡ Performance: LazyList e Cython

LazyList: Processamento com Memória O(1)

Baseado em factories de geradores para encadeamento sem materialização.

from koruspy import LazyList

# Pipeline complexo sem criar listas intermediárias
resultado = (LazyList(lambda: open('huge_file.txt'))
             .lazy_map(str.strip)
             .lazy_filter(lambda line: line.startswith('ERROR'))
             .lazy_map(lambda line: line.split('|'))
             .take(100)  # Pega apenas os 100 primeiros
             .to_list())

# Processamento infinito
fibonacci = (LazyList.iterate((0, 1), lambda pair: (pair[1], pair[0] + pair[1]))
             .lazy_map(lambda pair: pair[0])
             .take(10)
             .to_list())  # [0, 1, 1, 2, 3, 5, 8, 13, 21, 34]

Métodos principais:

• lazy_map(fn): Transformação preguiçosa
• lazy_filter(predicate): Filtragem preguiçosa
• take(n): Limita quantidade de elementos
• for_each(fn): Consome o iterador aplicando função
• to_list(): Materializa para lista Python

println: Motor de Saída em Cython

Impressão otimizada via C-API com type checking nativo.

from koruspy import println

# Uso direto
println("Hello, World!")  # Mais rápido que print()

# Integração com LazyList
(LazyList(lambda: range(100_000))
 .lazy_filter(lambda x: x % 2 == 0)
 .for_each(println))  # ~3.7x mais rápido que print()

Performance: Utiliza printf diretamente ao invés do mecanismo de IO do Python.

SomeList: Lista Avançada com Option Integration

from koruspy import SomeList

nums = SomeList([1, 2, 3, 4, 5])

# Operações seguras
total = nums.sum()  # Option[int]
doubled = nums.map(lambda x: x * 2)

# Conversão tipada
integers = SomeList(["1", "2", "foo", "3"]).to_integerlist()
# SomeList([1, 2, 3]) - invalidos são filtrados

# Imutabilidade
frozen_list = nums.freeze()  # Torna-se read-only

---

🌐 Programação Assíncrona: AsyncOption

Extensão do padrão Option para o mundo asyncio.

from koruspy import async_option, AsyncOption

async def fetch_user_score(user_id: int) -> AsyncOption[int]:
    user = await async_option(api.get_user(user_id))
    
    return (user
            .map_async(lambda u: u.profile)
            .filter_async(lambda p: p.verified)
            .map_async(lambda p: p.game_score)
            .filter_async(lambda s: s > 1000))

# Uso
score = await fetch_user_score(42).unwrap_or_async(0)

Métodos principais:

• map_async(fn): Transformação assíncrona
• filter_async(predicate): Filtragem assíncrona
• and_then_async(fn): Encadeamento de AsyncOptions
• unwrap_or_async(default): Resolução final com fallback

Vantagem: Constrói o pipeline inteiro antes de executar, evitando await prematuro.

---

🏗️ Filosofia de Design

Imutabilidade Opcional

from koruspy import Some

value = Some(42).freeze()  # Torna-se imutável
# value.map(...) # ❌ Erro: objeto frozen

Zero Memory Overhead

Todas as classes principais utilizam __slots__ para evitar o overhead do __dict__:

# Koruspy internals
class Some:
    __slots__ = ('_value',)

Fluent API

Inspirada em Scala e Rust para código autodescritivo:

(option_of(user)
 .filter(lambda u: u.age >= 18)
 .map(lambda u: u.email)
 .map(str.lower)
 .unwrap_or("[e-mail]"))

---

📊 Comparativo: Python Tradicional vs Koruspy

Aspecto	Python Tradicional	Koruspy
None Safety	if x is not None: repetitivo	Option elimina checagens
Error Handling	try-except aninhados	Result como valor
Memory (1M items)	~38 MB (listas materializadas)	~0.5 MB (lazy evaluation)
IO Performance	print() padrão	println() 3.7x mais rápido
Async Composition	await em cada passo	Pipeline único com finalize

---

🎓 Casos de Uso

1. ETL de Dados

from koruspy import LazyList, result_of

(LazyList(lambda: open('data.csv'))
 .lazy_map(str.strip)
 .lazy_map(lambda line: line.split(','))
 .lazy_filter(lambda cols: len(cols) == 5)
 .lazy_map(lambda cols: result_of(lambda: int(cols[2])))
 .lazy_filter(lambda r: r.is_ok())
 .lazy_map(lambda r: r.unwrap())
 .for_each(println))

2. API com Validação em Camadas

async def create_user(data: dict):
    return await (async_option(validate_email(data.get('email')))
                  .and_then_async(lambda e: check_unique(e))
                  .map_async(lambda e: save_user(data))
                  .unwrap_or_async(ErrorResponse("Invalid data")))

3. Análise de Logs em Real-Time

(LazyList(lambda: tail_file('/var/log/app.log'))
 .lazy_filter(lambda line: 'ERROR' in line)
 .lazy_map(parse_log_entry)
 .lazy_filter(lambda entry: entry.severity > 3)
 .for_each(send_alert))

---

📄 Licença

MIT License - veja LICENSE para detalhes.

---

🔗 Links & contacts

• gmail: [e-mail]
• PyPI: pypi.org/project/koruspy

---

Koruspy: Escreva Python como se fosse Rust. Performance de C. Elegância de Scala. 🚀

English 🇺🇸🇬🇧

Koruspy

Type safety, extreme performance, and elegant code for Python.

Koruspy is a functional utilities library that brings battle-tested patterns from Rust and Scala to the Python ecosystem. It eliminates NoneType errors and unhandled exceptions through Monads (Option and Result), while delivering C-level performance via Cython and lazy data processing with O(1) memory consumption.

---

🎯 Why Koruspy?

The library solves three critical Python development problems:

1. Robustness: Eliminate NoneType errors and silent exceptions

# ❌ Traditional Python: defensive and verbose code
def get_user_email(user_id):
    user = db.find_user(user_id)
    if user is not None:
        if hasattr(user, 'profile'):
            if user.profile is not None:
                return user.profile.email.lower()
    return "[e-mail]"

# ✅ Koruspy: safe, concise, and readable
def get_user_email(user_id):
    return (option_of(db.find_user(user_id))
            .and_then(lambda u: option_of(u.profile))
            .getattr("email")
            .map(str.lower)
            .unwrap_or("[e-mail]"))

2. Performance: Lazy processing + Cython engine

# ❌ Traditional Python: memory overflow and slow IO
logs = [i for i in range(1_000_000)]  # Entire list in RAM
filtered = [log * 2 for log in logs if log % 10 == 0]
for item in filtered:
    print(item)  # Native print() is slow at high volume

# ✅ Koruspy: Stable O(1) RAM and ultra-fast printing
from koruspy import LazyList, println

(LazyList(lambda: range(1_000_000))
 .lazy_filter(lambda x: x % 10 == 0)
 .lazy_map(lambda x: x * 2)
 .for_each(println))  # Cython engine with direct printf

Benchmark: Process and print 100k items in ~0.87s (vs ~3.2s with native print()).

3. Ergonomics: Fluent API that reads like natural language

# ✅ Async chaining without intermediate awaits
async def get_active_profile(user_id):
    return await (async_option(fetch_user(user_id))
                  .map_async(lambda u: u.profile)
                  .filter_async(lambda p: p.is_active)
                  .map_async(lambda p: p.to_dict())
                  .unwrap_or_async({"status": "inactive"}))

---

📦 Installation

pip install koruspy

---

🔥 Core: Option and Result Pattern

Option: Presence or Absence of Value

Replaces manual None checks with a semantic container.

from koruspy import option_of, Some, nothing

# Creation
user = option_of(db.get_user(123))  # Some(User) or nothing

# Safe transformation
email = (user
         .map(lambda u: u.email)
         .map(str.upper)
         .unwrap_or("[e-mail]"))

# Filtering
active_user = user.filter(lambda u: u.is_active)

# Conditional chaining
profile_pic = user.and_then(lambda u: option_of(u.profile_url))

Main methods:

• map(fn): Transforms the inner value if it exists
• and_then(fn): Chains operations that return Option
• filter(predicate): Keeps the value only if predicate is true
• unwrap_or(default): Extracts the value or returns default
• getattr(name): Safe attribute access

Visual debugging: Some appears in green, nothing in red in the terminal.

Result: Success or Failure

Transforms exceptions into treatable values.

from koruspy import result_of, Okay, Err

# Automatic exception capture
def divide(a, b):
    return result_of(lambda: a / b)

result = divide(10, 2)  # Okay(5.0)
error = divide(10, 0)   # Err(ZeroDivisionError)

# Processing pipeline
processed = (divide(100, 5)
             .map(lambda x: x * 2)
             .flat_map(lambda x: divide(x, 4))
             .unwrap_or(0))  # 10.0

# Explicit matching
match result:
    case Okay(value):
        print(f"Success: {value}")
    case Err(error):
        print(f"Error: {error}")

Main methods:

• map(fn): Transforms the value on success
• flat_map(fn): Chains operations that return Result
• unwrap_or(default): Extracts the value or returns default on error
• is_ok() / is_err(): State checking

Visual debugging: Okay in green, Err in red with formatted stack trace.

---

⚡ Performance: LazyList and Cython

LazyList: Processing with O(1) Memory

Based on generator factories for chaining without materialization.

from koruspy import LazyList

# Complex pipeline without creating intermediate lists
result = (LazyList(lambda: open('huge_file.txt'))
          .lazy_map(str.strip)
          .lazy_filter(lambda line: line.startswith('ERROR'))
          .lazy_map(lambda line: line.split('|'))
          .take(100)  # Takes only the first 100
          .to_list())

# Infinite processing
fibonacci = (LazyList.iterate((0, 1), lambda pair: (pair[1], pair[0] + pair[1]))
             .lazy_map(lambda pair: pair[0])
             .take(10)
             .to_list())  # [0, 1, 1, 2, 3, 5, 8, 13, 21, 34]

Main methods:

• lazy_map(fn): Lazy transformation
• lazy_filter(predicate): Lazy filtering
• take(n): Limits number of elements
• for_each(fn): Consumes the iterator applying function
• to_list(): Materializes to Python list

println: Cython Output Engine

Optimized printing via C-API with native type checking.

from koruspy import println

# Direct usage
println("Hello, World!")  # Faster than print()

# Integration with LazyList
(LazyList(lambda: range(100_000))
 .lazy_filter(lambda x: x % 2 == 0)
 .for_each(println))  # ~3.7x faster than print()

Performance: Uses printf directly instead of Python's IO mechanism.

SomeList: Advanced List with Option Integration

from koruspy import SomeList

nums = SomeList([1, 2, 3, 4, 5])

# Safe operations
total = nums.sum()  # Option[int]
doubled = nums.map(lambda x: x * 2)

# Typed conversion
integers = SomeList(["1", "2", "foo", "3"]).to_integerlist()
# SomeList([1, 2, 3]) - invalid values are filtered

# Immutability
frozen_list = nums.freeze()  # Becomes read-only

---

🌐 Async Programming: AsyncOption

Extension of the Option pattern to the asyncio world.

from koruspy import async_option, AsyncOption

async def fetch_user_score(user_id: int) -> AsyncOption[int]:
    user = await async_option(api.get_user(user_id))
    
    return (user
            .map_async(lambda u: u.profile)
            .filter_async(lambda p: p.verified)
            .map_async(lambda p: p.game_score)
            .filter_async(lambda s: s > 1000))

# Usage
score = await fetch_user_score(42).unwrap_or_async(0)

Main methods:

• map_async(fn): Async transformation
• filter_async(predicate): Async filtering
• and_then_async(fn): AsyncOption chaining
• unwrap_or_async(default): Final resolution with fallback

Advantage: Builds the entire pipeline before executing, avoiding premature await.

---

🏗️ Design Philosophy

Optional Immutability

from koruspy import Some

value = Some(42).freeze()  # Becomes immutable
# value.map(...) # ❌ Error: frozen object

Zero Memory Overhead

All main classes use __slots__ to avoid __dict__ overhead:

# Koruspy internals
class Some:
    __slots__ = ('_value',)

Fluent API

Inspired by Scala and Rust for self-documenting code:

(option_of(user)
 .filter(lambda u: u.age >= 18)
 .map(lambda u: u.email)
 .map(str.lower)
 .unwrap_or("[e-mail]"))

---

📊 Comparison: Traditional Python vs Koruspy

Aspect	Traditional Python	Koruspy
None Safety	Repetitive if x is not None:	Option eliminates checks
Error Handling	Nested try-except	Result as value
Memory (1M items)	~38 MB (materialized lists)	~0.5 MB (lazy evaluation)
IO Performance	Standard print()	println() 3.7x faster
Async Composition	await at each step	Single pipeline with finalize

---

🎓 Use Cases

1. Data ETL

from koruspy import LazyList, result_of

(LazyList(lambda: open('data.csv'))
 .lazy_map(str.strip)
 .lazy_map(lambda line: line.split(','))
 .lazy_filter(lambda cols: len(cols) == 5)
 .lazy_map(lambda cols: result_of(lambda: int(cols[2])))
 .lazy_filter(lambda r: r.is_ok())
 .lazy_map(lambda r: r.unwrap())
 .for_each(println))

2. API with Layered Validation

async def create_user(data: dict):
    return await (async_option(validate_email(data.get('email')))
                  .and_then_async(lambda e: check_unique(e))
                  .map_async(lambda e: save_user(data))
                  .unwrap_or_async(ErrorResponse("Invalid data")))

3. Real-Time Log Analysis

(LazyList(lambda: tail_file('/var/log/app.log'))
 .lazy_filter(lambda line: 'ERROR' in line)
 .lazy_map(parse_log_entry)
 .lazy_filter(lambda entry: entry.severity > 3)
 .for_each(send_alert))

---

📄 License

MIT License - see LICENSE for details.

---

🔗 Links & contacts

• gmail: "[e-mail]"
• PyPI: pypi.org/project/koruspy

---

Koruspy: Write Python like Rust. C performance. Scala elegance. 🚀