Small improvements in scoping and type assertions

Author: Dekker1

src/minizinc/instance.py

@@ -236,14 +236,14 @@ async def solve_async(
 
         """
         status = Status.UNKNOWN
-        solution = None
         statistics: Dict[str, Any] = {}
 
         multiple_solutions = (
             all_solutions or intermediate_solutions or nr_solutions is not None
         )
-        if multiple_solutions:
-            solution = []
+        solution: Union[Optional[Any], List[Any]] = (
+            [] if multiple_solutions else None
+        )
 
         async for result in self.solutions(
             time_limit=time_limit,
@@ -261,7 +261,7 @@ async def solve_async(
             statistics.update(result.statistics)
             if result.solution is not None:
                 if multiple_solutions:
-                    assert solution is not None
+                    assert isinstance(solution, list)
                     solution.append(result.solution)
                 else:
                     solution = result.solution
@@ -302,12 +302,12 @@ async def diverse_solutions(
                 "mzn-analyse executable could not be located"
             )
 
-        try:
-            # Create a temporary file in which the diversity model (generated by mzn-analyse) is placed
-            div_file = tempfile.NamedTemporaryFile(
-                prefix="mzn_div", suffix=".mzn", delete=False
-            )
+        # Create a temporary file in which the diversity model (generated by mzn-analyse) is placed
+        div_file = tempfile.NamedTemporaryFile(
+            prefix="mzn_div", suffix=".mzn", delete=False
+        )
 
+        try:
             # Extract the diversity annotations.
             with self.files() as files:
                 div_anns = mzn_analyse.run(
@@ -610,12 +610,11 @@ async def solutions(
 
             # Run the MiniZinc process
             proc = await self._driver._create_process(cmd, solver=solver)
-            try:
-                assert isinstance(proc.stderr, asyncio.StreamReader)
-                assert isinstance(proc.stdout, asyncio.StreamReader)
-
-                read_stderr = asyncio.create_task(_read_all(proc.stderr))
+            assert isinstance(proc.stderr, asyncio.StreamReader)
+            assert isinstance(proc.stdout, asyncio.StreamReader)
+            read_stderr = asyncio.create_task(_read_all(proc.stderr))
 
+            try:
                 async for obj in decode_async_json_stream(
                     proc.stdout, cls=MZNJSONDecoder, enum_map=self._enum_map
                 ):
@@ -850,6 +849,7 @@ def analyse(self):
             if obj["type"] == "interface":
                 interface = obj
                 break
+        assert interface is not None
         old_method = self._method_cache
         self._method_cache = Method.from_string(interface["method"])
         self._input_cache = {}
@@ -1033,6 +1033,7 @@ def _parse_stream_obj(self, obj, statistics):
             if "_checker" in statistics:
                 tmp["_checker"] = statistics.pop("_checker")
 
+            assert self.output_type is not None
             solution = self.output_type(**tmp)
             statistics["time"] = timedelta(milliseconds=obj["time"])
         elif obj["type"] == "time":