Stdin/stdout support (#161)

* Split _fstringify_file to fstringify_content

* Add stdin/stdout support

Fixes #160

* Update src/flynt/cli.py

Co-authored-by: Ilya Kamen <ikamenshchikov@gmail.com>

README.md

@@ -43,7 +43,8 @@ usage: flynt [-h] [-v | -q] [--no-multiline | -ll LINE_LENGTH] [-d]
 flynt v.0.77-beta
 
 positional arguments:
-  src                   source file(s) or directory
+  src                   source file(s) or directory (or a single `-`
+                        to read stdin and output to stdout)
 
 options:
   -h, --help            show this help message and exit

src/flynt/api.py

@@ -1,5 +1,6 @@
 import ast
 import codecs
+import dataclasses
 import logging
 import os
 import sys
@@ -22,33 +23,63 @@
 blacklist = {".tox", "venv", "site-packages", ".eggs"}
 
 
+@dataclasses.dataclass(frozen=True)
+class FstringifyResult:
+    n_changes: int
+    original_length: int
+    new_length: int
+    content: str
+
+
 def _fstringify_file(
     filename: str,
     state: State,
-) -> Tuple[bool, int, int, int]:
+) -> Optional[FstringifyResult]:
     """
-    :return: tuple: (changes_made, n_changes,
-    length of original code, length of new code)
+    F-stringify a file, write changes, and return a change result.
     """
-
-    def default_result() -> Tuple[bool, int, int, int]:
-        return False, 0, len(contents), len(contents)
-
     encoding, bom = encoding_by_bom(filename)
 
     with open(filename, encoding=encoding, newline="") as f:
         try:
             contents = f.read()
         except UnicodeDecodeError:
-            contents = ""
             log.error(f"Exception while reading {filename}", exc_info=True)
-            return default_result()
+            return None
+
+    result = fstringify_content(
+        contents=contents,
+        state=state,
+        filename=filename,
+    )
+
+    if result and result.n_changes:  # success?
+        new_code = result.content
+        if state.dry_run:
+            diff = unified_diff(
+                contents.split("\n"),
+                new_code.split("\n"),
+                fromfile=filename,
+            )
+            print("\n".join(diff))
+        else:
+            with open(filename, "wb") as outf:
+                if bom is not None:
+                    outf.write(bom)
+                outf.write(new_code.encode(encoding))
+    return result
 
+
+def fstringify_content(
+    contents: str,
+    state: State,
+    filename: str = "<code>",
+) -> Optional[FstringifyResult]:
     try:
         ast_before = ast.parse(contents)
     except SyntaxError:
         log.exception(f"Can't parse {filename} as a python file.")
-        return default_result()
+        return None
 
     try:
         new_code = contents
@@ -79,10 +110,17 @@ def default_result() -> Tuple[bool, int, int, int]:
             f"Skipping fstrings transform of file {filename} due to {msg}.",
             exc_info=True,
         )
-        return default_result()
+        return None
 
-    if new_code == contents:
-        return default_result()
+    result = FstringifyResult(
+        n_changes=changes,
+        original_length=len(contents),
+        new_length=len(new_code),
+        content=new_code,
+    )
+
+    if result.content == contents:
+        return result
 
     try:
         ast_after = ast.parse(new_code)
@@ -91,29 +129,15 @@ def default_result() -> Tuple[bool, int, int, int]:
             f"Faulty result during conversion on {filename} - skipping.",
             exc_info=True,
         )
-        return default_result()
+        return None
 
     if not len(ast_before.body) == len(ast_after.body):
         log.error(
             f"Faulty result during conversion on {filename}: "
             f"statement count has changed, which is not intended - skipping.",
         )
-        return default_result()
-
-    if state.dry_run:
-        diff = unified_diff(
-            contents.split("\n"),
-            new_code.split("\n"),
-            fromfile=filename,
-        )
-        print("\n".join(diff))
-    else:
-        with open(filename, "wb") as outf:
-            if bom is not None:
-                outf.write(bom)
-            outf.write(new_code.encode(encoding))
-
-    return True, changes, len(contents), len(new_code)
+        return None
+    return result
 
 
 def fstringify_files(
@@ -126,22 +150,19 @@ def fstringify_files(
     total_expressions = 0
     start_time = time.time()
     for path in files:
-        (
-            changed,
-            count_expressions,
-            charcount_original,
-            charcount_new,
-        ) = _fstringify_file(
+        result = _fstringify_file(
             path,
             state,
         )
-        if changed:
-            changed_files += 1
-            total_expressions += count_expressions
-        total_charcount_original += charcount_original
-        total_charcount_new += charcount_new
-
-        status = "modified" if count_expressions else "no change"
+        if result:
+            if result.n_changes:
+                changed_files += 1
+                total_expressions += result.n_changes
+            total_charcount_original += result.original_length
+            total_charcount_new += result.n_changes
+            status = "modified" if result.n_changes else "no change"
+        else:
+            status = "failed"
         log.info(f"fstringifying {path}...{status}")
     total_time = time.time() - start_time
 

src/flynt/cli.py

@@ -7,7 +7,7 @@
 from typing import List, Optional
 
 from flynt import __version__
-from flynt.api import fstringify, fstringify_code_by_line
+from flynt.api import fstringify, fstringify_code_by_line, fstringify_content
 from flynt.pyproject_finder import find_pyproject_toml, parse_pyproject_toml
 from flynt.state import State
 
@@ -140,7 +140,10 @@ def run_flynt_cli(arglist: Optional[List[str]] = None) -> int:
         "src",
         action="store",
         nargs="*",
-        help="source file(s) or directory",
+        help=(
+            "source file(s) or directory "
+            "(or a single `-` to read stdin and output to stdout)"
+        ),
     )
 
     parser.add_argument(
@@ -191,7 +194,18 @@ def run_flynt_cli(arglist: Optional[List[str]] = None) -> int:
         )
         print(converted)
         return 0
-
+    if "-" in args.src:
+        if len(args.src) > 1:
+            parser.error("Cannot use '-' with a list of other paths")
+        result = fstringify_content(
+            sys.stdin.read(),
+            state,
+            filename="<stdin>",
+        )
+        if not result:
+            return 1
+        print(result.content)
+        return 0
     salutation = f"Running flynt v.{__version__}"
     toml_file = find_pyproject_toml(tuple(args.src))
     if toml_file:

test/integration/test_api.py

@@ -61,26 +61,24 @@ def test_py2(py2_file):
     with open(py2_file) as f:
         content_before = f.read()
 
-    modified, _, _, _ = _fstringify_file(
-        py2_file, state=State(multiline=True, len_limit=1000)
-    )
+    result = _fstringify_file(py2_file, state=State(multiline=True, len_limit=1000))
 
     with open(py2_file) as f:
         content_after = f.read()
 
-    assert not modified
+    assert not result
     assert content_after == content_before
 
 
 def test_invalid_unicode(invalid_unicode_file):
-    modified, _, _, _ = _fstringify_file(
+    result = _fstringify_file(
         invalid_unicode_file, state=State(multiline=True, len_limit=1000)
     )
 
     with open(invalid_unicode_file, "rb") as f:
         content_after = f.read()
 
-    assert not modified
+    assert not result
     assert content_after == invalid_unicode
 
 
@@ -89,14 +87,14 @@ def test_works(formattable_file):
     with open(formattable_file) as f:
         content_before = f.read()
 
-    modified, _, _, _ = _fstringify_file(
+    result = _fstringify_file(
         formattable_file, state=State(multiline=True, len_limit=1000)
     )
 
     with open(formattable_file) as f:
         content_after = f.read()
 
-    assert modified
+    assert result.n_changes
     assert content_after != content_before
 
 
@@ -110,14 +108,14 @@ def broken_fstringify_by_line(*args, **kwargs):
 
     monkeypatch.setattr(api, "fstringify_code_by_line", broken_fstringify_by_line)
 
-    modified, _, _, _ = _fstringify_file(
+    result = _fstringify_file(
         formattable_file, state=State(multiline=True, len_limit=1000)
     )
 
     with open(formattable_file) as f:
         content_after = f.read()
 
-    assert not modified
+    assert not result
     assert content_after == content_before
 
 
@@ -131,41 +129,41 @@ def broken_fstringify_by_line(*args, **kwargs):
 
     monkeypatch.setattr(api, "fstringify_code_by_line", broken_fstringify_by_line)
 
-    modified, _, _, _ = _fstringify_file(
+    result = _fstringify_file(
         formattable_file, state=State(multiline=True, len_limit=1000)
     )
 
     with open(formattable_file) as f:
         content_after = f.read()
 
-    assert not modified
+    assert not result
     assert content_after == content_before
 
 
 def test_dry_run(formattable_file, monkeypatch):
     with open(formattable_file) as f:
         content_before = f.read()
 
-    modified, _, _, _ = _fstringify_file(
+    result = _fstringify_file(
         formattable_file, state=State(multiline=True, len_limit=1000, dry_run=True)
     )
 
     with open(formattable_file) as f:
         content_after = f.read()
 
-    assert modified
+    assert result.n_changes
     assert content_after == content_before
 
 
 def test_mixed_line_endings(mixed_line_endings_file):
-    modified, _, _, _ = _fstringify_file(
+    result = _fstringify_file(
         mixed_line_endings_file, state=State(multiline=True, len_limit=1000)
     )
 
     with open(mixed_line_endings_file, "rb") as f:
         content_after = f.read()
 
-    assert modified
+    assert result.n_changes
     assert content_after == mixed_line_endings_after
 
 
@@ -184,7 +182,5 @@ def test_bom(bom_file):
 
     It's possible to verify that a file has bom using `file` unix utility."""
 
-    modified, _, _, _ = _fstringify_file(
-        bom_file, state=State(multiline=True, len_limit=1000)
-    )
-    assert modified
+    result = _fstringify_file(bom_file, state=State(multiline=True, len_limit=1000))
+    assert result.n_changes