|
30 | 30 | cast,
|
31 | 31 | )
|
32 | 32 |
|
33 |
| -from schema_salad.utils import json_dumps |
34 |
| -from typing_extensions import TYPE_CHECKING, TypedDict |
35 |
| - |
36 | 33 | import prov.model as provM
|
37 | 34 | from prov.model import PROV, ProvDocument
|
| 35 | +from schema_salad.utils import json_dumps |
| 36 | +from typing_extensions import TYPE_CHECKING, TypedDict |
38 | 37 |
|
39 | 38 | from .loghandler import _logger
|
40 | 39 | from .provenance_constants import (
|
|
78 | 77 | pass
|
79 | 78 |
|
80 | 79 | if TYPE_CHECKING:
|
81 |
| - from .command_line_tool import ( |
| 80 | + from .command_line_tool import ( # pylint: disable=unused-import |
82 | 81 | CommandLineTool,
|
83 | 82 | ExpressionTool,
|
84 |
| - ) # pylint: disable=unused-import |
| 83 | + ) |
85 | 84 | from .workflow import Workflow # pylint: disable=unused-import
|
86 | 85 |
|
87 | 86 |
|
@@ -128,8 +127,8 @@ def __init__(self, research_object: "ResearchObject", rel_path: str) -> None:
|
128 | 127 | _logger.debug("[provenance] Creating WritableBagFile at %s.", path)
|
129 | 128 | super(WritableBagFile, self).__init__(path, mode="w")
|
130 | 129 |
|
131 |
| - def write(self, b: Union[bytes, str]) -> int: |
132 |
| - real_b = b if isinstance(b, bytes) else b.encode("utf-8") |
| 130 | + def write(self, b: Any) -> int: |
| 131 | + real_b = b if isinstance(b, (bytes, mmap, array)) else b.encode("utf-8") |
133 | 132 | total = 0
|
134 | 133 | length = len(real_b)
|
135 | 134 | while total < length:
|
|
0 commit comments