Merge branch 'main' into fix/flatten-command-type

Author: JSCU-CNI

.git-blame-ignore-revs

@@ -0,0 +1,6 @@
+# Formatting commits. You can ignore them during git-blame with `--ignore-rev` or `--ignore-revs-file`.
+#
+#   $ git config --add 'blame.ignoreRevsFile' '.git-blame-ignore-revs'
+#
+# Change linter to Ruff (#158)
+c67f778c653c295ec26146cf6422d3b06ac640e8

examples/__init__.py

@@ -1,10 +1,15 @@
-import os
-import stat
+from __future__ import annotations
 
-from datetime import datetime
+import stat
+from pathlib import Path
+from typing import TYPE_CHECKING
 
 from flow.record import RecordDescriptor, RecordWriter
 
+if TYPE_CHECKING:
+    from collections.abc import Iterator
+
+
 descriptor = """
 filesystem/unix/entry
     string path;
@@ -22,34 +27,32 @@
 FilesystemFile = RecordDescriptor(descriptor)
 
 
-def hash_file(path, t):
-    f = open(path, "rb")
-    while 1:
-        d = f.read(4096)
-        if d == "":
-            break
-    f.close()
+def hash_file(path: str | Path) -> None:
+    with Path(path).open("rb") as f:
+        while True:
+            d = f.read(4096)
+            if not d:
+                break
 
 
 class FilesystemIterator:
     basepath = None
 
-    def __init__(self, basepath):
+    def __init__(self, basepath: str | None):
         self.basepath = basepath
         self.recordType = FilesystemFile
 
-    def classify(self, source, classification):
+    def classify(self, source: str, classification: str) -> None:
         self.recordType = FilesystemFile.base(_source=source, _classification=classification)
 
-    def iter(self, path):
-        path = os.path.abspath(path)
-        return self._iter(path)
+    def iter(self, path: str | Path) -> Iterator[FilesystemFile]:
+        return self._iter(Path(path).resolve())
 
-    def _iter(self, path):
-        if path.startswith("/proc"):
+    def _iter(self, path: Path) -> Iterator[FilesystemFile]:
+        if path.is_relative_to("/proc"):
             return
 
-        st = os.lstat(path)
+        st = path.lstat()
 
         abspath = path
         if self.basepath and abspath.startswith(self.basepath):
@@ -59,7 +62,7 @@ def _iter(self, path):
 
         link = None
         if ifmt == stat.S_IFLNK:
-            link = os.readlink(path)
+            link = path.readlink()
 
         yield self.recordType(
             path=abspath,
@@ -69,20 +72,16 @@ def _iter(self, path):
             size=st.st_size,
             uid=st.st_uid,
             gid=st.st_gid,
-            ctime=datetime.fromtimestamp(st.st_ctime),
-            mtime=datetime.fromtimestamp(st.st_mtime),
-            atime=datetime.fromtimestamp(st.st_atime),
+            ctime=st.st_ctime,
+            mtime=st.st_mtime,
+            atime=st.st_atime,
             link=link,
         )
 
         if ifmt == stat.S_IFDIR:
-            for i in os.listdir(path):
-                if i in (".", ".."):
-                    continue
-
-                fullpath = os.path.join(path, i)
-                for e in self.iter(fullpath):
-                    yield e
+            for i in path.iterdir():
+                fullpath = path.joinpath(i)
+                yield from self.iter(fullpath)
 
 
 chunk = []

examples/filesystem.py

@@ -1,18 +1,21 @@
 #!/usr/bin/env pypy
-import record
+from __future__ import annotations
+
 import sys
-import datetime
+from datetime import datetime, timezone
 
 import net.ipv4
-
+import record
 from fileprocessing import DirectoryProcessor
 
+UTC_TIMEZONE = timezone.utc
+
 
-def ts(s):
-    return datetime.datetime.fromtimestamp(float(s))
+def ts(s: float) -> datetime:
+    return datetime.fromtimestamp(float(s), tz=UTC_TIMEZONE)
 
 
-def ip(s):
+def ip(s: str) -> net.ipv4.Address:
     return net.ipv4.Address(s)
 
 
@@ -21,7 +24,7 @@ class SeparatedFile:
     seperator = None
     format = None
 
-    def __init__(self, fp, seperator, format):
+    def __init__(self, fp: list[str], seperator: str | None, format: list[tuple]):
         self.fp = fp
         self.seperator = seperator
         self.format = format
@@ -46,7 +49,7 @@ def __iter__(self):
             yield recordtype(**r)
 
 
-def PassiveDnsFile(fp):
+def PassiveDnsFile(fp: list[str]) -> SeparatedFile:
     return SeparatedFile(fp, "||", PASSIVEDNS_FORMAT)
 
 
@@ -63,7 +66,7 @@ def PassiveDnsFile(fp):
 ]
 
 
-def main():
+def main() -> None:
     rs = record.RecordOutput(sys.stdout)
     for r in DirectoryProcessor(sys.argv[1], PassiveDnsFile, r"\.log\.gz"):
         rs.write(r)

examples/passivedns.py

@@ -1,8 +1,10 @@
 import random
+from datetime import datetime, timezone
 
-from datetime import datetime
 from flow import record
 
+UTC_TIMEZONE = timezone.utc
+
 descriptor = """
 network/traffic/tcp/connection
     datetime ts;
@@ -32,9 +34,9 @@
 
 rs = record.RecordWriter()
 
-for i in range(500):
+for _ in range(500):
     r = conn(
-        ts=datetime.now(),
+        ts=datetime.now(tz=UTC_TIMEZONE),
         src=random.choice(ip_list),
         srcport=random.choice(port_list),
         dst=random.choice(ip_list),

examples/tcpconn.py

@@ -1,5 +1,7 @@
+from __future__ import annotations
+
 import gzip
-import os
+from pathlib import Path
 
 from flow.record.base import (
     IGNORE_FIELDS_FOR_COMPARISON,
@@ -39,71 +41,61 @@
 
 __all__ = [
     "IGNORE_FIELDS_FOR_COMPARISON",
-    "RECORD_VERSION",
     "RECORDSTREAM_MAGIC",
+    "RECORD_VERSION",
+    "DynamicDescriptor",
     "FieldType",
-    "Record",
     "GroupedRecord",
-    "RecordDescriptor",
+    "JsonRecordPacker",
+    "PathTemplateWriter",
+    "Record",
     "RecordAdapter",
+    "RecordArchiver",
+    "RecordDescriptor",
+    "RecordDescriptorError",
     "RecordField",
-    "RecordReader",
-    "RecordWriter",
     "RecordOutput",
-    "RecordPrinter",
     "RecordPacker",
-    "JsonRecordPacker",
-    "RecordStreamWriter",
+    "RecordPrinter",
+    "RecordReader",
     "RecordStreamReader",
-    "open_path_or_stream",
+    "RecordStreamWriter",
+    "RecordWriter",
+    "dynamic_fieldtype",
+    "extend_record",
+    "ignore_fields_for_comparison",
+    "iter_timestamped_records",
     "open_path",
+    "open_path_or_stream",
     "open_stream",
-    "ignore_fields_for_comparison",
+    "record_stream",
     "set_ignored_fields_for_comparison",
     "stream",
-    "dynamic_fieldtype",
-    "DynamicDescriptor",
-    "PathTemplateWriter",
-    "RecordArchiver",
-    "RecordDescriptorError",
-    "record_stream",
-    "extend_record",
-    "iter_timestamped_records",
 ]
 
 
-class View:
-    fields = None
-
-    def __init__(self, fields):
-        self.fields = fields
-
-    def __iter__(self, fields):
-        pass
-
-
 class RecordDateSplitter:
     basepath = None
     out = None
 
-    def __init__(self, basepath):
-        self.basepath = basepath
+    def __init__(self, basepath: str | Path):
+        self.basepath = Path(basepath)
         self.out = {}
 
-    def getstream(self, t):
+    def getstream(self, t: tuple[int, int, int]) -> RecordStreamWriter:
         if t not in self.out:
-            path = os.path.join(self.basepath, "-".join(["{:2d}".format(v) for v in t]) + ".rec.gz")
+            path = self.basepath.joinpath("-".join([f"{v:2d}" for v in t]) + ".rec.gz")
             f = gzip.GzipFile(path, "wb")
             rs = RecordStreamWriter(f)
             self.out[t] = rs
         return self.out[t]
 
-    def write(self, r):
+    def write(self, r: Record) -> None:
         t = (r.ts.year, r.ts.month, r.ts.day)
         rs = self.getstream(t)
         rs.write(r)
         rs.fp.flush()
 
-    def close(self):
+    def close(self) -> None:
         for rs in self.out.values():
             rs.close()

flow/record/__init__.py

@@ -1,63 +1,53 @@
+from __future__ import annotations
+
 __path__ = __import__("pkgutil").extend_path(__path__, __name__)  # make this namespace extensible from other packages
 import abc
+from typing import TYPE_CHECKING
 
+if TYPE_CHECKING:
+    from collections.abc import Iterator
 
-def with_metaclass(meta, *bases):
-    """Create a base class with a metaclass. Python 2 and 3 compatible."""
-
-    # This requires a bit of explanation: the basic idea is to make a dummy
-    # metaclass for one level of class instantiation that replaces itself with
-    # the actual metaclass.
-    class metaclass(type):
-        def __new__(cls, name, this_bases, d):
-            return meta(name, bases, d)
-
-        @classmethod
-        def __prepare__(cls, name, this_bases):
-            return meta.__prepare__(name, bases)
-
-    return type.__new__(metaclass, "temporary_class", (), {})
+    from flow.record.base import Record
 
 
-class AbstractWriter(with_metaclass(abc.ABCMeta, object)):
+class AbstractWriter(metaclass=abc.ABCMeta):
     @abc.abstractmethod
-    def write(self, rec):
+    def write(self, rec: Record) -> None:
         """Write a record."""
         raise NotImplementedError
 
     @abc.abstractmethod
-    def flush(self):
+    def flush(self) -> None:
         """Flush any buffered writes."""
         raise NotImplementedError
 
     @abc.abstractmethod
-    def close(self):
+    def close(self) -> None:
         """Close the Writer, no more writes will be possible."""
         raise NotImplementedError
 
-    def __del__(self):
+    def __del__(self) -> None:
         self.close()
 
-    def __enter__(self):
+    def __enter__(self) -> AbstractWriter:  # noqa: PYI034
         return self
 
-    def __exit__(self, *args):
+    def __exit__(self, *args) -> None:
         self.flush()
         self.close()
 
 
-class AbstractReader(with_metaclass(abc.ABCMeta, object)):
+class AbstractReader(metaclass=abc.ABCMeta):
     @abc.abstractmethod
-    def __iter__(self):
+    def __iter__(self) -> Iterator[Record]:
         """Return a record iterator."""
         raise NotImplementedError
 
-    def close(self):
+    def close(self) -> None:  # noqa: B027
         """Close the Reader, can be overriden to properly free resources."""
-        pass
 
-    def __enter__(self):
+    def __enter__(self) -> AbstractReader:  # noqa: PYI034
         return self
 
-    def __exit__(self, *args):
+    def __exit__(self, *args) -> None:
         self.close()