Upgrade to 0.0.9

Author: trisongz

changelogs.txt

@@ -1,3 +1,7 @@
+v0.0.9 - Aug 5, 2021
+- Modifications to LazyRPC
+- Lots of other fun stuff.
+
 v0.0.81 - July 28, 2021
 
 - Add API for LazyHFModel

lazyops/__init__.py

@@ -1,3 +1,4 @@
+from lazyops.cache.asyncish import lazy_async
 import os
 import warnings
 
@@ -13,8 +14,8 @@
 from .models import LazyData, LazyTime, LazyDate, LazyFormatter, LazyTimer, LazyObject
 from .common import lazylibs, lazy_init, run_cmd, clone_repo, File
 from .utils import find_binary_in_path, timed_cache, latest_tf_ckpt, require_module
-from .utils import build_chunks, retryable, list_to_dict
-from .mp import lazy_parallelize, lazyproc, lazymultiproc, LazyProcs, LazyProc
+from .utils import build_chunks, retryable, list_to_dict, create_uuid
+from .mp import lazy_parallelize, lazyproc, lazymultiproc, LazyProcs, LazyProc, async_to_sync
 from .apis import LazyAPI, LazyAPIConfig
 
 
@@ -29,6 +30,7 @@
 fio = File
 lazyapi = LazyAPI
 lazyapiconfig = LazyAPIConfig
+lazy_async = async_to_sync
 
 from .lazyio import LazyHFModel
 
@@ -65,6 +67,8 @@
     'lazyproc',
     'lazymultiproc',
     'LazyProcs',
-    'LazyProc'
+    'LazyProc',
+    'lazy_async',
+    'async_to_sync'
 ]
 #from . import lazyrpc

lazyops/envs.py

@@ -188,7 +188,7 @@ def exit_handler(cls, signum, frame):
             for thread in EnvChecker.threads:
                 thread.join()
         EnvChecker.set_dead()
-        #sys.exit(0)
+        sys.exit(0)
 
     @classmethod
     def enable_watcher(cls):

lazyops/lazyio/__init__.py

@@ -1,4 +1,5 @@
 from .models import (
+    LazyFile,
     LazyJson,
     LazyPickler,
     LazyIOBase,
@@ -8,14 +9,18 @@
     LazyIOPickle,
     LazyIOType,
     LazyHFModel,
+    gfile,
+    File,
+    json,
 )
 
 
 lazypickler = LazyPickler
 lazypkler = LazyPickler
 lazyjson = LazyJson
 
-lazyfile = LazyIOBase
+#lazyfile = LazyIOBase
+lazyfile = LazyFile
 lazyiojson = LazyIOJson
 lazyjsonlines = LazyIOJsonLines
 lazyjlines = LazyIOJsonLines
@@ -25,6 +30,7 @@
 
 
 __all__ = [
+    'LazyFile',
     'LazyJson',
     'LazyPickler',
     'LazyIOBase',
@@ -43,6 +49,9 @@
     'lazyjlines',
     'lazyjl',
     'lazypickle',
-    'lazypkl'
+    'lazypkl',
+    'File',
+    'gfile',
+    'json'
 ]
 

lazyops/lazyio/_base.py

@@ -5,9 +5,12 @@
 import simdjson as json
 import os
 
+from pathlib import Path
 from enum import Enum
 from abc import ABCMeta
 from typing import Union, List, Any, TypeVar, Optional, Dict
 from fileio import File, gfile
+from tempfile import TemporaryFile, NamedTemporaryFile
+from io import BytesIO
 
 logger = get_logger(name='LazyIO')

lazyops/lazyio/models.py

@@ -1,5 +1,45 @@
 from ._base import *
 
+# make stuff compatible and compliant with posix
+# https://www.tensorflow.org/datasets/api_docs/python/tfds/core/ReadWritePath?hl=en
+
+class LazyFile:
+    def __init__(self, filename, method='tmp', overwrite=True, cleanup=True):
+        self._dest_filename = filename
+        self._method = method
+        self.overwrite = overwrite
+        self.cleanup = cleanup
+        self._tmp_file = NamedTemporaryFile(delete=False) if method == 'tmp' else BytesIO()
+    
+    @property
+    def tmpfile(self):
+        return self._tmp_file
+    
+    @property
+    def tmpfile_path(self):
+        if self._tmp_file and self._method == 'tmp':
+            return self._tmp_file.name
+        return None
+    
+    @property
+    def is_closed(self):
+        return bool(self._tmp_file is not None)
+
+    def write(self):
+        assert not self.is_closed
+        assert File.exists(self._dest_filename) is False or self.overwrite is True
+        if self._method == 'tmp':
+            data = gfile(self._tmp_file.name, 'rb').read()
+        else:
+            data = self._tmp_file.getvalue()
+        with gfile(self._dest_filename, 'wb') as f:
+            f.write(data)
+        if self.cleanup and self._method == 'tmp':
+            os.remove(self._tmp_file.name)
+        self._tmp_file = None
+
+
+
 class LazyJson:
     serializer = json
     parser = json.Parser()

lazyops/lazyrpc/_base.py

@@ -15,7 +15,7 @@
 from json import JSONDecodeError
 from types import FunctionType, CoroutineType
 from typing import List, Union, Any, Callable, Type, Optional, Dict, Sequence, Awaitable, Tuple
-from contextlib import AsyncExitStack, AbstractAsyncContextManager, asynccontextmanager
+from contextlib import AsyncExitStack, AbstractAsyncContextManager, asynccontextmanager, contextmanager
 
 # noinspection PyProtectedMember
 from pydantic import DictError
@@ -65,6 +65,13 @@ def __get__(self, obj, cls):
             value = obj.__dict__[self.func.__name__] = self.func(obj)
             return value
 
+try:
+    import sentry_sdk
+    from sentry_sdk.utils import transaction_from_function as sentry_transaction_from_function
+except ImportError:
+    sentry_sdk = None
+    sentry_transaction_from_function = None
+
 def sjson_loads(data, *args, **kwargs):
     try:
         return _JSON_PARSER.parse(data, recursive=kwargs.get('recursive', False))

lazyops/lazyrpc/core.py

@@ -89,14 +89,16 @@ def __init__(
         http_request: Request,
         background_tasks: BackgroundTasks,
         http_response: Response,
-        json_rpc_request_class: Type[JsonRpcRequest] = JsonRpcRequest
+        json_rpc_request_class: Type[JsonRpcRequest] = JsonRpcRequest,
+        method_route: Optional['MethodRoute'] = None,
     ):
         self.entrypoint: Entrypoint = entrypoint
         self.raw_request: Any = raw_request
         self.http_request: Request = http_request
         self.background_tasks: BackgroundTasks = background_tasks
         self.http_response: Response = http_response
         self.request_class: Type[JsonRpcRequest] = json_rpc_request_class
+        self.method_route: Optional[MethodRoute] = method_route
         self._raw_response: Optional[dict] = None
         self.exception: Optional[Exception] = None
         self.is_unhandled_exception: bool = False
@@ -142,6 +144,8 @@ def request(self) -> JsonRpcRequest:
     async def __aenter__(self):
         assert self.exit_stack is None
         self.exit_stack = await AsyncExitStack().__aenter__()
+        if sentry_sdk is not None:
+            self.exit_stack.enter_context(self._fix_sentry_scope())
         await self.exit_stack.enter_async_context(self._handle_exception(reraise=False))
         self.jsonrpc_context_token = _jsonrpc_context.set(self)
         return self
@@ -174,6 +178,23 @@ async def _handle_exception(self, reraise=True):
         if self.exception is not None and self.is_unhandled_exception:
             logger.exception(str(self.exception), exc_info=self.exception)
 
+    def _make_sentry_event_processor(self):
+        def event_processor(event, _):
+            if self.method_route is not None:
+                event['transaction'] = sentry_transaction_from_function(self.method_route.func)
+            return event
+
+        return event_processor
+
+    @contextmanager
+    def _fix_sentry_scope(self):
+        hub = sentry_sdk.Hub.current
+        with sentry_sdk.Hub(hub) as hub:
+            with hub.configure_scope() as scope:
+                scope.clear_breadcrumbs()
+                scope.add_event_processor(self._make_sentry_event_processor())
+                yield
+
     async def enter_middlewares(self, middlewares: Sequence['JsonRpcMiddleware']):
         for mw in middlewares:
             cm = mw(self)
@@ -354,6 +375,7 @@ async def handle_req_to_resp(
     ) -> dict:
         async with JsonRpcContext(
             entrypoint=self.entrypoint,
+            method_route=self,
             raw_request=req,
             http_request=http_request,
             background_tasks=background_tasks,
@@ -714,6 +736,7 @@ async def handle_req(
         for route in self.entrypoint.routes:  # type: MethodRoute
             match, child_scope = route.matches(http_request_shadow.scope)
             if match == Match.FULL:
+                ctx.method_route = route
                 # http_request is a transport layer and it is common for all JSON-RPC requests in a batch
                 return await route.handle_req(
                     http_request_shadow, background_tasks, sub_response, ctx,
@@ -823,6 +846,7 @@ def add_method_route(
         **kwargs,
     ) -> None:
         name = name or func.__name__
+        #name = self.entrypoint_route.name + ' - ' + (name or func.__name__)
         route = self.method_route_class(
             self,
             self.entrypoint_route.path + '/' + name,

lazyops/lazyrpc/exceptions.py

@@ -1,5 +1,6 @@
 from ._base import *
 from .models import component_name, rename_if_scope_child_component
+from fastapi import status
 
 class BaseError(Exception):
     CODE = None
@@ -158,33 +159,38 @@ class ErrorModel(BaseModel):
 
 class ParseError(BaseError):
     """Invalid JSON was received by the server"""
-    CODE = -32700
+    CODE = status.HTTP_422_UNPROCESSABLE_ENTITY
+    #CODE = -32700
     MESSAGE = "Parse error"
 
 
 class InvalidRequest(BaseError):
     """The JSON sent is not a valid Request object"""
-    CODE = -32600
+    #CODE = -32600
+    CODE = status.HTTP_400_BAD_REQUEST
     MESSAGE = "Invalid Request"
     error_model = ErrorModel
 
 
 class MethodNotFound(BaseError):
     """The method does not exist / is not available"""
-    CODE = -32601
+    #CODE = -32601
+    CODE = status.HTTP_405_METHOD_NOT_ALLOWED
     MESSAGE = "Method not found"
 
 
 class InvalidParams(BaseError):
     """Invalid method parameter(s)"""
-    CODE = -32602
+    #CODE = -32602
+    CODE = status.HTTP_424_FAILED_DEPENDENCY
     MESSAGE = "Invalid params"
     error_model = ErrorModel
 
 
 class InternalError(BaseError):
     """Internal JSON-RPC error"""
-    CODE = -32603
+    #CODE = -32603
+    CODE = status.HTTP_500_INTERNAL_SERVER_ERROR
     MESSAGE = "Internal error"
 
 
@@ -196,10 +202,19 @@ def errors_responses(errors: Sequence[Type[BaseError]] = None):
     if errors:
         cnt = 1
         for error_cls in errors:
-            responses[f'200{" " * cnt}'] = {
+            #responses[404] = {
+            #    'model': error_cls.get_resp_model(),
+            #    'description': error_cls.get_description(),
+            #}
+            responses[error_cls.CODE] = {
                 'model': error_cls.get_resp_model(),
                 'description': error_cls.get_description(),
             }
+            
+            #responses[f'200{" " * cnt}'] = {
+            #    'model': error_cls.get_resp_model(),
+            #    'description': error_cls.get_description(),
+            #}
             cnt += 1
     return responses
 

lazyops/mp.py

@@ -1,6 +1,7 @@
 
 from functools import lru_cache, wraps
 from typing import List, Optional, Dict
+from asgiref.sync import async_to_sync
 from .mp_utils import multiproc, _MAX_PROCS, lazy_parallelize
 
 class LazyProc: