RxPY
v4.0.4
Python에서 관찰 가능한 컬렉션과 쿼리 연산자 함수를 사용하여 비동기 및 이벤트 기반 프로그램을 구성하기 위한 라이브러리
v3.X의 경우 v3 분기로 이동하세요.
Python v4.x용 ReactiveX는 Python 3.7 이상에서 실행됩니다. 설치하려면:
pip3 install reactivex ReactiveX for Python(RxPY)은 Python에서 관찰 가능한 시퀀스와 파이프 가능한 쿼리 연산자를 사용하여 비동기 및 이벤트 기반 프로그램을 구성하기 위한 라이브러리입니다. Rx를 사용하여 개발자는 Observable로 비동기 데이터 스트림을 표현하고, 연산자를 사용하여 비동기 데이터 스트림을 쿼리하고, 스케줄러를 사용하여 데이터/이벤트 스트림의 동시성을 매개변수화합니다.
import reactivex as rx
from reactivex import operators as ops
source = rx . of ( "Alpha" , "Beta" , "Gamma" , "Delta" , "Epsilon" )
composed = source . pipe (
ops . map ( lambda s : len ( s )),
ops . filter ( lambda i : i >= 5 )
)
composed . subscribe ( lambda value : print ( "Received {0}" . format ( value )))ReactiveX의 원리를 배우고 사용 가능한 연산자에 대한 전체 참조를 얻으려면 설명서를 읽어보세요.
RxPY v1.x 또는 v3.x에서 코드를 마이그레이션해야 하는 경우 마이그레이션 섹션을 읽어보세요.
여기에는 제3자 문서 목록도 있습니다.
GitHub 토론에서 대화에 참여하세요! 질문이나 제안 사항이 있는 경우.
Python용 ReactiveX는 120개 이상의 연산자와 1300개 이상의 통과 단위 테스트를 갖춘 상당히 완전한 Rx 구현입니다. RxPY는 대부분 RxJS의 직접 포트이지만 스레딩 및 차단 연산자 측면에서 Rx.NET 및 RxJava의 일부도 차용했습니다.
Python용 ReactiveX는 PEP 8을 따르므로 모든 함수 및 메서드 이름은 snake_cased 됩니다. 즉, 가독성을 높이기 위해 필요에 따라 단어를 밑줄로 구분한 소문자입니다.
따라서 다음과 같은 .NET 코드:
var group = source . GroupBy ( i => i % 3 ) ; Python에서는 _ 를 사용하여 작성해야 합니다.
group = source . pipe ( ops . group_by ( lambda i : i % 3 ))Python용 ReactiveX를 사용하면 연산자에 여러 개의 선택적 인수가 있는 경우 위치 인수 대신 명명된 키워드 인수를 사용해야 합니다. RxPY는 연산자에 어떤 인수를 제공하는지(또는 제공하지 않는지) 감지하려고 시도하지 않습니다.
이 프로젝트는 Poetry를 사용하여 관리됩니다. 코드는 Black, isort를 사용하여 형식화됩니다. 코드는 pyright 및 mypy를 사용하여 정적으로 유형 검사됩니다.
기본 VSCode 통합을 활용하려면 먼저 Poetry를 구성하여 저장소에 가상 환경을 만드세요.
poetry config virtualenvs.in-project true저장소를 복제한 후 도구를 활성화합니다.
poetry install
poetry run pre-commit install단위 테스트를 실행합니다.
poetry run pytest코드 검사 실행(수동):
poetry run pre-commit run --all-files
