首页>编程相关>其他源码
下载

༄ streamable

类似溪流的迭代操作

Stream[T]用流利的界面装饰了一个Iterable[T]从而使懒惰操作的链接。

UNITSEST打字皮棉PYPI

？流利链方法！
？打字类型通知和mypy
？懒惰的在迭代时对操作进行了懒惰的评估
并发通过线程或进程或asyncio
？对Python 3.7至3.14的单位测试，覆盖率为100％
？极简主义者pip install streamable无需其他依赖项

1。安装

pip install streamable

2。导入

 from streamable import Stream

3。初始化

从一个Iterable[T]实例化Stream[T] 。 

 integers : Stream [ int ] = Stream ( range ( 10 ))

4。操作

  • Stream S是不变的：应用操作返回新流。

  • 操作是懒惰的：仅在迭代时进行评估。 

 inverses : Stream [ float ] = (
    integers
    . map ( lambda n : round ( 1 / n , 2 ))
    . catch ( ZeroDivisionError )
)

5。迭代

  • 像您在任何其他Iterable[T]上一样，在Stream[T]上迭代。
  • 源元素是在即时处理的。

收集它

 >> > list ( inverses )
[ 1.0 , 0.5 , 0.33 , 0.25 , 0.2 , 0.17 , 0.14 , 0.12 , 0.11 ]
>> > set ( inverses )
{ 0.5 , 1.0 , 0.2 , 0.33 , 0.25 , 0.17 , 0.14 , 0.12 , 0.11 }

减少它

 >> > sum ( inverses )
2.82
>> > max ( inverses )
1.0
>> > from functools import reduce
>> > reduce (..., inverses )

循环

 >> > for inverse in inverses :
>> >    ...

接下来

 >> > inverses_iter = iter ( inverses )
>> > next ( inverses_iter )
1.0
>> > next ( inverses_iter )
0.5

？运营

.map

对元素进行转换： 

 negative_integer_strings : Stream [ str ] = integers . map ( lambda n : - n ). map ( str )

assert list ( negative_integer_strings ) == [ '0' , '-1' , '-2' , '-3' , '-4' , '-5' , '-6' , '-7' , '-8' , '-9' ]

基于线程的并发

通过concurrency线程应用转换： 

 import requests

pokemon_names : Stream [ str ] = (
    Stream ( range ( 1 , 4 ))
    . map ( lambda i : f"https://pokeapi.co/api/v2/pokemon-species/ { i } " )
    . map ( requests . get , concurrency = 3 )
    . map ( requests . Response . json )
    . map ( lambda poke : poke [ "name" ])
)
assert list ( pokemon_names ) == [ 'bulbasaur' , 'ivysaur' , 'venusaur' ]

默认情况下保留上游订单（FIFO），但是您可以将第一个完成的ordered=False设置为false。

concurrency也是含有尚未取消结果的缓冲区的大小。如果缓冲区已满，则暂停上游的迭代，直到从缓冲区产生结果为止。

基于过程的并发

via="process"设置： 

 if __name__ == "__main__" :
    state : List [ int ] = []
    n_integers : int = integers . map ( state . append , concurrency = 4 , via = "process" ). count ()
    assert n_integers == 10
    assert state == [] # main process's state not mutated

基于异步的并发

同胞操作.amap应用异步函数： 

 import httpx
import asyncio

http_async_client = httpx . AsyncClient ()

pokemon_names : Stream [ str ] = (
    Stream ( range ( 1 , 4 ))
    . map ( lambda i : f"https://pokeapi.co/api/v2/pokemon-species/ { i } " )
    . amap ( http_async_client . get , concurrency = 3 )
    . map ( httpx . Response . json )
    . map ( lambda poke : poke [ "name" ])
)

assert list ( pokemon_names ) == [ 'bulbasaur' , 'ivysaur' , 'venusaur' ]
asyncio . get_event_loop (). run_until_complete ( http_async_client . aclose ())

starmap

star函数装饰器会转换一个将几个位置参数的函数转换为带有元组的函数： 

 from streamable import star

zeros : Stream [ int ] = (
    Stream ( enumerate ( integers ))
    . map ( star ( lambda index , integer : index - integer ))
)

assert list ( zeros ) == [ 0 , 0 , 0 , 0 , 0 , 0 , 0 , 0 , 0 , 0 ]

.filter很.foreach 。

.foreach

对元素应用副作用： 

 self_printing_integers : Stream [ int ] = integers . foreach ( print )

assert list ( self_printing_integers ) == list ( integers )  # triggers the printing

基于线程的并发

像.map一样，它具有可选的concurrency参数。

基于过程的并发

喜欢.map ， via="process"设置参数。

基于异步的并发

就像.map一样，它具有异步的同级.aforeach操作。

.filter

只保留满足条件的要素： 

 pair_integers : Stream [ int ] = integers . filter ( lambda n : n % 2 == 0 )

assert list ( pair_integers ) == [ 0 , 2 , 4 , 6 , 8 ]

.throttle

限制per_second / per_minute / per_hour的收益率数： 

 slow_integers : Stream [ int ] = integers . throttle ( per_second = 5 )

assert list ( slow_integers ) == list ( integers )  # takes 10 / 5 = 2 seconds

和/或确保最小时间interval连续的收益率： 

 from datetime import timedelta

slow_integers = integers . throttle ( interval = timedelta ( milliseconds = 100 ))

assert list ( slow_integers ) == list ( integers )  # takes 10 * 0.1 = 1 second

.group

将元素分组到List中： 

 integers_5_by_5 : Stream [ List [ int ]] = integers . group ( size = 5 )

assert list ( integers_5_by_5 ) == [[ 0 , 1 , 2 , 3 , 4 ], [ 5 , 6 , 7 , 8 , 9 ]]
 integers_by_parity : Stream [ List [ int ]] = integers . group ( by = lambda n : n % 2 )

assert list ( integers_by_parity ) == [[ 0 , 2 , 4 , 6 , 8 ], [ 1 , 3 , 5 , 7 , 9 ]]
 from datetime import timedelta

integers_within_1s : Stream [ List [ int ]] = (
    integers
    . throttle ( per_second = 2 )
    . group ( interval = timedelta ( seconds = 0.99 ))
)

assert list ( integers_within_1s ) == [[ 0 , 1 , 2 ], [ 3 , 4 ], [ 5 , 6 ], [ 7 , 8 ], [ 9 ]]

混合size / by / interval参数： 

 integers_2_by_2_by_parity : Stream [ List [ int ]] = integers . group ( by = lambda n : n % 2 , size = 2 )

assert list ( integers_2_by_2_by_parity ) == [[ 0 , 2 ], [ 1 , 3 ], [ 4 , 6 ], [ 5 , 7 ], [ 8 ], [ 9 ]]

.flatten

假设它们是Iterable 。 

 pair_then_odd_integers : Stream [ int ] = integers_by_parity . flatten ()

assert list ( pair_then_odd_integers ) == [ 0 , 2 , 4 , 6 , 8 , 1 , 3 , 5 , 7 , 9 ]

基于线程的并发

同时concurrency并发迭代： 

 mix_of_0s_and_1s : Stream [ int ] = Stream ([[ 0 ] * 4 , [ 1 ] * 4 ]). flatten ( concurrency = 2 )
assert list ( mix_of_0s_and_1s ) == [ 0 , 1 , 0 , 1 , 0 , 1 , 0 , 1 ]

.catch

捕获给定类型的异常，并可选地产生replacement值： 

 inverses : Stream [ float ] = (
    integers
    . map ( lambda n : round ( 1 / n , 2 ))
    . catch ( ZeroDivisionError , replacement = float ( "inf" ))
)

assert list ( inverses ) == [ float ( "inf" ), 1.0 , 0.5 , 0.33 , 0.25 , 0.2 , 0.17 , 0.14 , 0.12 , 0.11 ]

您可以when捕获条件下指定其他： 

 import requests
from requests . exceptions import SSLError

status_codes_ignoring_resolution_errors : Stream [ int ] = (
    Stream ([ "https://github.com" , "https://foo.bar" , "https://github.com/foo/bar" ])
    . map ( requests . get , concurrency = 2 )
    . catch ( SSLError , when = lambda exception : "Max retries exceeded with url" in str ( exception ))
    . map ( lambda response : response . status_code )
)

assert list ( status_codes_ignoring_resolution_errors ) == [ 200 , 404 ]

它具有可选的finally_raise: bool参数，以提高迭代结束时第一个捕获的异常。

.truncate

一旦产生给定数量的元素，结束迭代： 

 five_first_integers : Stream [ int ] = integers . truncate ( 5 )

assert list ( five_first_integers ) == [ 0 , 1 , 2 , 3 , 4 ]

...或当条件得到满足时： 

 five_first_integers : Stream [ int ] = integers . truncate ( when = lambda n : n == 5 )

assert list ( five_first_integers ) == [ 0 , 1 , 2 , 3 , 4 ]

.observe

如果您迭代： 

 observed_slow_integers : Stream [ int ] = slow_integers . observe ( "integers" )

您将获得这些日志： 

 INFO: [duration=0:00:00.502155 errors=0] 1 integers yielded
INFO: [duration=0:00:01.006336 errors=0] 2 integers yielded
INFO: [duration=0:00:02.011921 errors=0] 4 integers yielded
INFO: [duration=0:00:04.029666 errors=0] 8 integers yielded
INFO: [duration=0:00:05.039571 errors=0] 10 integers yielded

日志的量永远不会是压倒性的，因为它们是对数产生的，例如，当迭代到达第1024个元素时，将产生第11 log。

警告

它是v1.1.0和v1.3.1之间pip install --upgrade streamable静音

zip

使用标准zip功能： 

 from streamable import star

cubes : Stream [ int ] = (
    Stream ( zip ( integers , integers , integers )) # Stream[Tuple[int, int, int]]
    . map ( star ( lambda a , b , c : a * b * c ))
)

assert list ( cubes ) == [ 0 , 1 , 8 , 27 , 64 , 125 , 216 , 343 , 512 , 729 ]

？笔记框

贡献

请帮助我们！非常欢迎：

  • 开放问题
  • 打开拉的请求
  • 检查贡献

耗尽流

.count 

 >> > assert integers . count () == 10

调用流在其上迭代直到耗尽并返回。 

 >> > verbose_integers : Stream [ int ] = integers . foreach ( print )
>> > assert verbose_integers () is verbose_integers
0
1
2
3
4
5
6
7
8
9

提取 - 转换负载

ETL脚本（即脚本获取 - >处理 - >推送数据）可以从该库的表达性中受益。

这是一个示例，您可以复制和尝试（仅需要requests ）：它创建一个CSV文件，其中包含来自第一，第二和第3代Pokémons（kudos topokéapi）的所有67个四倍体

 import csv
from datetime import timedelta
import itertools
import requests
from streamable import Stream

with open ( "./quadruped_pokemons.csv" , mode = "w" ) as file :
    fields = [ "id" , "name" , "is_legendary" , "base_happiness" , "capture_rate" ]
    writer = csv . DictWriter ( file , fields , extrasaction = 'ignore' )
    writer . writeheader ()
    (
        # Infinite Stream[int] of Pokemon ids starting from Pokémon #1: Bulbasaur
        Stream ( itertools . count ( 1 ))
        # Limits to 16 requests per second to be friendly to our fellow PokéAPI devs
        . throttle ( per_second = 16 )
        # GETs pokemons concurrently using a pool of 8 threads
        . map ( lambda poke_id : f"https://pokeapi.co/api/v2/pokemon-species/ { poke_id } " )
        . map ( requests . get , concurrency = 8 )
        . foreach ( requests . Response . raise_for_status )
        . map ( requests . Response . json )
        # Stops the iteration when reaching the 1st pokemon of the 4th generation
        . truncate ( when = lambda poke : poke [ "generation" ][ "name" ] == "generation-iv" )
        . observe ( "pokemons" )
        # Keeps only quadruped Pokemons
        . filter ( lambda poke : poke [ "shape" ][ "name" ] == "quadruped" )
        . observe ( "quadruped pokemons" )
        # Catches errors due to None "generation" or "shape"
        . catch (
            TypeError ,
            when = lambda error : str ( error ) == "'NoneType' object is not subscriptable"
        )
        # Writes a batch of pokemons every 5 seconds to the CSV file
        . group ( interval = timedelta ( seconds = 5 ))
        . foreach ( writer . writerows )
        . flatten ()
        . observe ( "written pokemons" )
        # Catches exceptions and raises the 1st one at the end of the iteration
        . catch ( finally_raise = True )
        # Actually triggers an iteration (the lines above define lazy operations)
        . count ()
    )

记录级别

 logging . getLogger ( "streamable" ). setLevel ( logging . WARNING )  # default is INFO

访客模式

Stream类公开了.accept方法，您可以通过扩展streamable.visitors.Visitor摘要类来实现访客： 

 from streamable . visitors import Visitor

class DepthVisitor ( Visitor [ int ]):
    def visit_stream ( self , stream : Stream ) -> int :
        if not stream . upstream :
            return 1
        return 1 + stream . upstream . accept ( self )

def depth ( stream : Stream ) -> int :
    return stream . accept ( DepthVisitor ())

assert depth ( Stream ( range ( 10 )). map ( str ). filter ()) == 3

作为功​​能

Stream的方法也被视为功能： 

 from streamable . functions import catch

inverse_integers : Iterator [ int ] = map ( lambda n : 1 / n , range ( 10 ))
safe_inverse_integers : Iterator [ int ] = catch ( inverse_integers , ZeroDivisionError )

与自由线python兼容3.13+

从自由线python的3.13+构建中受益，通过python -X gil=0运行。

Ty荧光笔

  • 在reddit上高投资器
  • @pythonhub
  • pycoders.com
  • planetpython.org
展开
附加信息
相关应用
为您推荐
相关资讯 全部