linter and type checker

src/ocsf/compare/__main__.py

@@ -23,8 +23,12 @@
 def main():
     parser = ArgumentParser(description="Compare two OCSF schemata")
 
-    parser.add_argument("old_schema", help="Path to the old schema file, old schema repository, or the old schema version.")
-    parser.add_argument("new_schema", help="Path to the new schema file, new schema repository, or the new schema version.")
+    parser.add_argument(
+        "old_schema", help="Path to the old schema file, old schema repository, or the old schema version."
+    )
+    parser.add_argument(
+        "new_schema", help="Path to the new schema file, new schema repository, or the new schema version."
+    )
     parser.add_argument(
         "--expand-changes",
         dest="collapse_changes",

src/ocsf/compile/__main__.py

@@ -49,6 +49,7 @@
 from .compiler import Compilation
 from .options import CompilationOptions
 
+
 def main():
     parser = ArgumentParser(description="Compile an OCSF repository into a schema and dump it as JSON to STDOUT")
     parser.add_argument("path", help="Path to the OCSF repository")
@@ -119,5 +120,6 @@ def main():
 
     print(to_json(compiler.build()))
 
+
 if __name__ == "__main__":
-    main()
+    main()

src/ocsf/compile/planners/observable.py

@@ -43,6 +43,7 @@
 
 # TODO build observable object's type_id enum
 
+
 class _Registry:
     """A registry of observable attributes and types from the dictionary.json
     file so that the mapping can be computed once across all operations.
@@ -122,6 +123,7 @@ def apply(self, schema: ProtoSchema) -> MergeResult:
 
         return results
 
+
 @dataclass(eq=True, frozen=True)
 class BuildObservableTypeOp(Operation):
     registry: Optional[_Registry] = None
@@ -155,7 +157,9 @@ def apply(self, schema: ProtoSchema) -> MergeResult:
                 enum_id = str(attrs[key])
                 attr = data.attributes[key]
                 assert isinstance(attr, AttrDefn)
-                enum[enum_id] = EnumMemberDefn(caption=attr.caption, description=f"Observable by Dictionary Attribute.<br>{attr.description}")
+                enum[enum_id] = EnumMemberDefn(
+                    caption=attr.caption, description=f"Observable by Dictionary Attribute.<br>{attr.description}"
+                )
                 results.append(("attributes", "type_id", "enum", enum_id))
 
             # Dictionary type observables
@@ -167,9 +171,10 @@ def apply(self, schema: ProtoSchema) -> MergeResult:
                 enum_id = str(types[key])
                 type_ = data.types.attributes[key]
                 assert isinstance(type_, TypeDefn)
-                enum[enum_id] = EnumMemberDefn(caption=type_.caption, description=f"Observable by Dictionary Type.<br>{type_.description}")
+                enum[enum_id] = EnumMemberDefn(
+                    caption=type_.caption, description=f"Observable by Dictionary Type.<br>{type_.description}"
+                )
                 results.append(("attributes", "type_id", "enum", enum_id))
-
 
         elif isinstance(data, EventDefn) or isinstance(data, ObjectDefn):
             if isinstance(data, ObjectDefn) and data.observable is not None:
@@ -184,7 +189,9 @@ def apply(self, schema: ProtoSchema) -> MergeResult:
                 # Object observable
                 enum_id = str(obj_data.observable)
                 if enum_id not in enum:
-                    enum[enum_id] = EnumMemberDefn(caption=obj_data.caption, description=f"Observable by Object.<br>{obj_data.description}")
+                    enum[enum_id] = EnumMemberDefn(
+                        caption=obj_data.caption, description=f"Observable by Object.<br>{obj_data.description}"
+                    )
                     results.append(("attributes", "type_id", "enum", enum_id))
 
             if isinstance(data.attributes, dict):
@@ -193,14 +200,16 @@ def apply(self, schema: ProtoSchema) -> MergeResult:
                 for k, v in data.attributes.items():
                     if isinstance(v, AttrDefn) and v.observable is not None:
                         enum_id = str(v.observable)
-                        if enum_id not in enum: # Don't overwrite enum values defined in dictionary.json
-                            enum[enum_id] = EnumMemberDefn(caption=f"{data.caption} {label}: {k}", description=f"Observable by {label}-Specific Attribute.<br>{label}-specific attribute \"{k}\" for the {data.caption} {label}.")
+                        if enum_id not in enum:  # Don't overwrite enum values defined in dictionary.json
+                            enum[enum_id] = EnumMemberDefn(
+                                caption=f"{data.caption} {label}: {k}",
+                                description=f'Observable by {label}-Specific Attribute.<br>{label}-specific attribute "{k}" for the {data.caption} {label}.',
+                            )
                             results.append(("attributes", "type_id", "enum", enum_id))
 
         return results
 
 
-
 class MarkObservablesPlanner(Planner):
     def __init__(self, schema: ProtoSchema, options: CompilationOptions):
         super().__init__(schema, options)
@@ -213,7 +222,7 @@ def analyze(self, input: DefinitionFile[AnyDefinition]) -> Analysis:
             ops.append(MarkObservablesOp(input.path, registry=self._registry))
 
         return ops
-            
+
 
 class BuildObservableTypesPlanner(Planner):
     def __init__(self, schema: ProtoSchema, options: CompilationOptions):
@@ -223,7 +232,11 @@ def __init__(self, schema: ProtoSchema, options: CompilationOptions):
     def analyze(self, input: DefinitionFile[AnyDefinition]) -> Analysis:
         ops: Analysis = []
 
-        if input.path == SpecialFiles.DICTIONARY or (isinstance(input.data, ObjectDefn) or isinstance(input.data, EventDefn)):
-            ops.append(BuildObservableTypeOp(target=SpecialFiles.OBSERVABLE, prerequisite=input.path, registry=self._registry))
-
-        return ops
+        if input.path == SpecialFiles.DICTIONARY or (
+            isinstance(input.data, ObjectDefn) or isinstance(input.data, EventDefn)
+        ):
+            ops.append(
+                BuildObservableTypeOp(target=SpecialFiles.OBSERVABLE, prerequisite=input.path, registry=self._registry)
+            )
+
+        return ops

src/ocsf/compile/protoschema.py

@@ -1,15 +1,13 @@
 import dacite
 
 from copy import deepcopy
-from dataclasses import asdict, dataclass
+from dataclasses import asdict
 from pathlib import PurePath
 from typing import Any, cast, TypeVar
-from enum import IntEnum
 
 from ocsf.schema import OcsfSchema, OcsfObject, OcsfEvent, OcsfType, OcsfProfile, OcsfExtension
 from ocsf.repository import (
     Repository,
-    DefnWithExtends,
     ObjectDefn,
     EventDefn,
     SpecialFiles,

src/ocsf/repository/definitions.py

@@ -230,4 +230,4 @@ class CategoriesDefn(DefinitionData):
 This is only used for definitions that create new record types in the core schema. `dictionary.json` is exempt.
 """
 
-DefnWithExtends = ObjectDefn | EventDefn
+DefnWithExtends = ObjectDefn | EventDefn

src/ocsf/schema/__main__.py

@@ -17,7 +17,10 @@
 def main():
     parser = ArgumentParser(description="Dump an OCSF schema as JSON to STDOUT")
 
-    parser.add_argument("schema", help="Path to a schema JSON file, a version identifier (to retrieve from schema.ocsf.io), or a path to an OCSF repository.")
+    parser.add_argument(
+        "schema",
+        help="Path to a schema JSON file, a version identifier (to retrieve from schema.ocsf.io), or a path to an OCSF repository.",
+    )
     args = parser.parse_args()
 
     print(to_json(get_schema(args.schema)))

tests/ocsf/compile/planners/test_observable.py

@@ -10,7 +10,7 @@
 from ocsf.compile import CompilationOptions
 from ocsf.compile.protoschema import ProtoSchema
 from ocsf.compile.planners.observable import MarkObservablesPlanner, MarkObservablesOp
-from ocsf.compile.planners.observable import _Registry # type: ignore
+from ocsf.compile.planners.observable import _Registry  # type: ignore
 
 
 def get_ps():