Fix UP038 (isinstance(..., X | Y))

Author: Armavica

pymc/backends/arviz.py

@@ -78,7 +78,7 @@ def is_data(name, var, model) -> bool:
             and var not in model.potentials
             and var not in model.value_vars
             and name not in observations
-            and isinstance(var, (Constant, SharedVariable))
+            and isinstance(var, Constant | SharedVariable)
         )
 
     # The assumption is that constants (like pm.Data) are named

pymc/distributions/censored.py

@@ -89,7 +89,7 @@ class Censored(Distribution):
     @classmethod
     def dist(cls, dist, lower, upper, **kwargs):
         if not isinstance(dist, TensorVariable) or not isinstance(
-            dist.owner.op, (RandomVariable, SymbolicRandomVariable)
+            dist.owner.op, RandomVariable | SymbolicRandomVariable
         ):
             raise ValueError(
                 f"Censoring dist must be a distribution created via the `.dist()` API, got {type(dist)}"

pymc/distributions/distribution.py

@@ -102,7 +102,7 @@ def rewrite_support_point_scan_node(self, node):
 
         for nd in local_fgraph_topo:
             if nd not in to_replace_set and isinstance(
-                nd.op, (RandomVariable, SymbolicRandomVariable)
+                nd.op, RandomVariable | SymbolicRandomVariable
             ):
                 replace_with_support_point.append(nd.out)
                 to_replace_set.add(nd)
@@ -132,7 +132,7 @@ def add_requirements(self, fgraph):
 
     def apply(self, fgraph):
         for node in fgraph.toposort():
-            if isinstance(node.op, (RandomVariable, SymbolicRandomVariable)):
+            if isinstance(node.op, RandomVariable | SymbolicRandomVariable):
                 fgraph.replace(node.out, support_point(node.out))
             elif isinstance(node.op, Scan):
                 new_node = self.rewrite_support_point_scan_node(node)
@@ -837,7 +837,7 @@ def custom_dist_get_support_point(op, rv, size, *params):
                     *[
                         p
                         for p in params
-                        if not isinstance(p.type, (RandomType, RandomGeneratorType))
+                        if not isinstance(p.type, RandomType | RandomGeneratorType)
                     ],
                 )
 

pymc/distributions/mixture.py

@@ -178,7 +178,7 @@ class Mixture(Distribution):
 
     @classmethod
     def dist(cls, w, comp_dists, **kwargs):
-        if not isinstance(comp_dists, (tuple, list)):
+        if not isinstance(comp_dists, tuple | list):
             # comp_dists is a single component
             comp_dists = [comp_dists]
         elif len(comp_dists) == 1:
@@ -204,7 +204,7 @@ def dist(cls, w, comp_dists, **kwargs):
             # TODO: Allow these to not be a RandomVariable as long as we can call `ndim_supp` on them
             #  and resize them
             if not isinstance(dist, TensorVariable) or not isinstance(
-                dist.owner.op, (RandomVariable, SymbolicRandomVariable)
+                dist.owner.op, RandomVariable | SymbolicRandomVariable
             ):
                 raise ValueError(
                     f"Component dist must be a distribution created via the `.dist()` API, got {type(dist)}"

pymc/distributions/multivariate.py

@@ -1083,7 +1083,7 @@ def WishartBartlett(name, S, nu, is_cholesky=False, return_cholesky=False, initv
 
 def _lkj_normalizing_constant(eta, n):
     # TODO: This is mixing python branching with the potentially symbolic n and eta variables
-    if not isinstance(eta, (int, float)):
+    if not isinstance(eta, int | float):
         raise NotImplementedError("eta must be an int or float")
     if not isinstance(n, int):
         raise NotImplementedError("n must be an integer")
@@ -1185,7 +1185,7 @@ def dist(cls, n, eta, sd_dist, **kwargs):
         if not (
             isinstance(sd_dist, Variable)
             and sd_dist.owner is not None
-            and isinstance(sd_dist.owner.op, (RandomVariable, SymbolicRandomVariable))
+            and isinstance(sd_dist.owner.op, RandomVariable | SymbolicRandomVariable)
             and sd_dist.owner.op.ndim_supp < 2
         ):
             raise TypeError("sd_dist must be a scalar or vector distribution variable")
@@ -2262,7 +2262,7 @@ def logp(value, mu, W, alpha, tau):
         TensorVariable
         """
 
-        sparse = isinstance(W, (pytensor.sparse.SparseConstant, pytensor.sparse.SparseVariable))
+        sparse = isinstance(W, pytensor.sparse.SparseConstant | pytensor.sparse.SparseVariable)
 
         if sparse:
             D = sp_sum(W, axis=0)

pymc/distributions/shape_utils.py

@@ -193,7 +193,7 @@ def convert_dims(dims: Dims | None) -> StrongDims | None:
 
     if isinstance(dims, str):
         dims = (dims,)
-    elif isinstance(dims, (list, tuple)):
+    elif isinstance(dims, list | tuple):
         dims = tuple(dims)
     else:
         raise ValueError(f"The `dims` parameter must be a tuple, str or list. Actual: {type(dims)}")
@@ -209,7 +209,7 @@ def convert_shape(shape: Shape) -> StrongShape | None:
         shape = (shape,)
     elif isinstance(shape, TensorVariable) and shape.ndim == 1:
         shape = tuple(shape)
-    elif isinstance(shape, (list, tuple)):
+    elif isinstance(shape, list | tuple):
         shape = tuple(shape)
     else:
         raise ValueError(
@@ -227,7 +227,7 @@ def convert_size(size: Size) -> StrongSize | None:
         size = (size,)
     elif isinstance(size, TensorVariable) and size.ndim == 1:
         size = tuple(size)
-    elif isinstance(size, (list, tuple)):
+    elif isinstance(size, list | tuple):
         size = tuple(size)
     else:
         raise ValueError(

pymc/distributions/timeseries.py

@@ -88,15 +88,15 @@ def dist(cls, init_dist, innovation_dist, steps=None, **kwargs) -> pt.TensorVari
         if not (
             isinstance(init_dist, pt.TensorVariable)
             and init_dist.owner is not None
-            and isinstance(init_dist.owner.op, (RandomVariable, SymbolicRandomVariable))
+            and isinstance(init_dist.owner.op, RandomVariable | SymbolicRandomVariable)
         ):
             raise TypeError("init_dist must be a distribution variable")
         check_dist_not_registered(init_dist)
 
         if not (
             isinstance(innovation_dist, pt.TensorVariable)
             and innovation_dist.owner is not None
-            and isinstance(innovation_dist.owner.op, (RandomVariable, SymbolicRandomVariable))
+            and isinstance(innovation_dist.owner.op, RandomVariable | SymbolicRandomVariable)
         ):
             raise TypeError("innovation_dist must be a distribution variable")
         check_dist_not_registered(innovation_dist)
@@ -129,7 +129,7 @@ def get_steps(cls, innovation_dist, steps, shape, dims, observed):
         if not (
             isinstance(innovation_dist, pt.TensorVariable)
             and innovation_dist.owner is not None
-            and isinstance(innovation_dist.owner.op, (RandomVariable, SymbolicRandomVariable))
+            and isinstance(innovation_dist.owner.op, RandomVariable | SymbolicRandomVariable)
         ):
             raise TypeError("innovation_dist must be a distribution variable")
 
@@ -549,7 +549,7 @@ def dist(
 
         if init_dist is not None:
             if not isinstance(init_dist, TensorVariable) or not isinstance(
-                init_dist.owner.op, (RandomVariable, SymbolicRandomVariable)
+                init_dist.owner.op, RandomVariable | SymbolicRandomVariable
             ):
                 raise ValueError(
                     f"Init dist must be a distribution created via the `.dist()` API, "
@@ -948,7 +948,7 @@ def dist(cls, dt, sde_fn, sde_pars, *, init_dist=None, steps=None, **kwargs):
 
         if init_dist is not None:
             if not isinstance(init_dist, TensorVariable) or not isinstance(
-                init_dist.owner.op, (RandomVariable, SymbolicRandomVariable)
+                init_dist.owner.op, RandomVariable | SymbolicRandomVariable
             ):
                 raise ValueError(
                     f"Init dist must be a distribution created via the `.dist()` API, "

pymc/gp/cov.py

@@ -256,11 +256,7 @@ def _merge_factors_cov(self, X, Xs=None, diag=False):
 
             elif isinstance(
                 factor,
-                (
-                    TensorConstant,
-                    TensorVariable,
-                    TensorSharedVariable,
-                ),
+                TensorConstant | TensorVariable | TensorSharedVariable,
             ):
                 if factor.ndim == 2 and diag:
                     factor_list.append(pt.diag(factor))
@@ -524,7 +520,7 @@ def __init__(
         if (ls is None and ls_inv is None) or (ls is not None and ls_inv is not None):
             raise ValueError("Only one of 'ls' or 'ls_inv' must be provided")
         elif ls_inv is not None:
-            if isinstance(ls_inv, (list, tuple)):
+            if isinstance(ls_inv, list | tuple):
                 ls = 1.0 / np.asarray(ls_inv)
             else:
                 ls = 1.0 / ls_inv

pymc/gp/hsgp_approx.py

@@ -328,7 +328,7 @@ def prior_linearized(self, Xs: TensorLike):
 
         # If not provided, use Xs and c to set L
         if self._L is None:
-            assert isinstance(self._c, (numbers.Real, np.ndarray, pt.TensorVariable))
+            assert isinstance(self._c, numbers.Real | np.ndarray | pt.TensorVariable)
             self.L = pt.as_tensor(set_boundary(Xs, self._c))
         else:
             self.L = self._L

pymc/gp/util.py

@@ -113,7 +113,7 @@ def kmeans_inducing_points(n_inducing, X, **kmeans_kwargs):
     # first whiten X
     if isinstance(X, TensorConstant):
         X = X.value
-    elif isinstance(X, (np.ndarray, tuple, list)):
+    elif isinstance(X, np.ndarray | tuple | list):
         X = np.asarray(X)
     else:
         raise TypeError(

pymc/logprob/abstract.py

@@ -71,7 +71,7 @@ def _logprob_helper(rv, *values, **kwargs):
     if (not name) and (len(values) == 1):
         name = values[0].name
     if name:
-        if isinstance(logprob, (list, tuple)):
+        if isinstance(logprob, list | tuple):
             for i, term in enumerate(logprob):
                 term.name = f"{name}_logprob.{i}"
         else:

pymc/logprob/basic.py

@@ -78,7 +78,7 @@ def _find_unallowed_rvs_in_graph(graph):
     return {
         rv
         for rv in rvs_in_graph(graph)
-        if not isinstance(rv.owner.op, (SimulatorRV, MinibatchIndexRV))
+        if not isinstance(rv.owner.op, SimulatorRV | MinibatchIndexRV)
     }
 
 
@@ -545,7 +545,7 @@ def conditional_logp(
             **kwargs,
         )
 
-        if not isinstance(q_logprob_vars, (list, tuple)):
+        if not isinstance(q_logprob_vars, list | tuple):
             q_logprob_vars = [q_logprob_vars]
 
         for q_value_var, q_logprob_var in zip(q_values, q_logprob_vars):

pymc/logprob/binary.py

@@ -102,9 +102,9 @@ def comparison_logprob(op, values, base_rv, operand, **kwargs):
 
     condn_exp = pt.eq(value, np.array(True))
 
-    if isinstance(op.scalar_op, (GT, GE)):
+    if isinstance(op.scalar_op, GT | GE):
         logprob = pt.switch(condn_exp, logccdf, logcdf)
-    elif isinstance(op.scalar_op, (LT, LE)):
+    elif isinstance(op.scalar_op, LT | LE):
         logprob = pt.switch(condn_exp, logcdf, logccdf)
     else:
         raise TypeError(f"Unsupported scalar_op {op.scalar_op}")

pymc/logprob/mixture.py

@@ -248,7 +248,7 @@ def get_stack_mixture_vars(
     joined_rvs = node.inputs[0]
 
     # First, make sure that it's some sort of concatenation
-    if not (joined_rvs.owner and isinstance(joined_rvs.owner.op, (MakeVector, Join))):
+    if not (joined_rvs.owner and isinstance(joined_rvs.owner.op, MakeVector | Join)):
         return None, None
 
     if isinstance(joined_rvs.owner.op, MakeVector):
@@ -284,7 +284,7 @@ def find_measurable_index_mixture(fgraph, node):
     mixing_indices = node.inputs[1:]
 
     # TODO: Add check / test case for Advanced Boolean indexing
-    if isinstance(node.op, (AdvancedSubtensor, AdvancedSubtensor1)):
+    if isinstance(node.op, AdvancedSubtensor | AdvancedSubtensor1):
         # We don't support (non-scalar) integer array indexing as it can pick repeated values,
         # but the Mixture logprob assumes all mixture values are independent
         if any(
@@ -298,7 +298,7 @@ def find_measurable_index_mixture(fgraph, node):
     mixture_rvs, join_axis = get_stack_mixture_vars(node)
 
     # We don't support symbolic join axis
-    if mixture_rvs is None or not isinstance(join_axis, (NoneTypeT, Constant)):
+    if mixture_rvs is None or not isinstance(join_axis, NoneTypeT | Constant):
         return None
 
     if rv_map_feature.request_measurable(mixture_rvs) != mixture_rvs:

pymc/logprob/transforms.py

@@ -779,7 +779,7 @@ class PowerTransform(Transform):
     name = "power"
 
     def __init__(self, power=None):
-        if not isinstance(power, (int, float)):
+        if not isinstance(power, int | float):
             raise TypeError(f"Power must be integer or float, got {type(power)}")
         if power == 0:
             raise ValueError("Power cannot be 0")

pymc/logprob/utils.py

@@ -147,7 +147,7 @@ def expand(r):
     return {
         node
         for node in walk(makeiter(vars), expand, False)
-        if node.owner and isinstance(node.owner.op, (RandomVariable, MeasurableVariable))
+        if node.owner and isinstance(node.owner.op, RandomVariable | MeasurableVariable)
     }
 
 

pymc/model/core.py

@@ -192,7 +192,7 @@ def resolve_type(c: type | str) -> type:
         assert cls is not None
         if isinstance(cls._context_class, str):
             cls._context_class = resolve_type(cls._context_class)
-        if not isinstance(cls._context_class, (str, type)):
+        if not isinstance(cls._context_class, str | type):
             raise ValueError(
                 f"Context class for {cls.__name__}, {cls._context_class}, is not of the right type"
             )
@@ -695,7 +695,7 @@ def logp(
         varlist: list[TensorVariable]
         if vars is None:
             varlist = self.free_RVs + self.observed_RVs + self.potentials
-        elif not isinstance(vars, (list, tuple)):
+        elif not isinstance(vars, list | tuple):
             varlist = [vars]
         else:
             varlist = cast(list[TensorVariable], vars)
@@ -770,7 +770,7 @@ def dlogp(
         if vars is None:
             value_vars = None
         else:
-            if not isinstance(vars, (list, tuple)):
+            if not isinstance(vars, list | tuple):
                 vars = [vars]
 
             value_vars = []
@@ -809,7 +809,7 @@ def d2logp(
         if vars is None:
             value_vars = None
         else:
-            if not isinstance(vars, (list, tuple)):
+            if not isinstance(vars, list | tuple):
                 vars = [vars]
 
             value_vars = []
@@ -998,7 +998,7 @@ def add_coord(
         if name in self.coords:
             if not np.array_equal(values, self.coords[name]):
                 raise ValueError(f"Duplicate and incompatible coordinate: {name}.")
-        if length is not None and not isinstance(length, (int, Variable)):
+        if length is not None and not isinstance(length, int | Variable):
             raise ValueError(
                 f"The `length` passed for the '{name}' coord must be an int, PyTensor Variable or None."
             )
@@ -1070,7 +1070,7 @@ def initial_point(self, random_seed: SeedSequenceSeed = None) -> dict[str, np.nd
 
     def set_initval(self, rv_var, initval):
         """Sets an initial value (strategy) for a random variable."""
-        if initval is not None and not isinstance(initval, (Variable, str)):
+        if initval is not None and not isinstance(initval, Variable | str):
             # Convert scalars or array-like inputs to ndarrays
             initval = rv_var.type.filter(initval)
 

pymc/model/fgraph.py

@@ -260,7 +260,7 @@ def fgraph_from_model(
     inverse_memo = {v: k for k, v in memo.items()}
     for var, model_var in replacements:
         if not inlined_views and (
-            model_var.owner and isinstance(model_var.owner.op, (ModelDeterministic, ModelNamed))
+            model_var.owner and isinstance(model_var.owner.op, ModelDeterministic | ModelNamed)
         ):
             # Ignore extra identity that will be removed at the end
             var = var.owner.inputs[0]

pymc/model/transform/basic.py

@@ -55,7 +55,7 @@ def prune_vars_detached_from_observed(model: Model) -> Model:
 
 
 def parse_vars(model: Model, vars: ModelVariable | Sequence[ModelVariable]) -> list[Variable]:
-    if isinstance(vars, (list, tuple)):
+    if isinstance(vars, list | tuple):
         vars_seq = vars
     else:
         vars_seq = (vars,)

pymc/model/transform/conditioning.py

@@ -106,7 +106,7 @@ def observe(
         model_var = memo[var]
 
         # Just a sanity check
-        assert isinstance(model_var.owner.op, (ModelFreeRV, ModelDeterministic))
+        assert isinstance(model_var.owner.op, ModelFreeRV | ModelDeterministic)
         assert model_var in fgraph.variables
 
         var = model_var.owner.inputs[0]

pymc/ode/ode.py

@@ -149,9 +149,9 @@ def make_node(self, y0, theta):
         return Apply(self, inputs, (states, sens))
 
     def __call__(self, y0, theta, return_sens=False, **kwargs):
-        if isinstance(y0, (list, tuple)) and not len(y0) == self.n_states:
+        if isinstance(y0, list | tuple) and not len(y0) == self.n_states:
             raise ShapeError("Length of y0 is wrong.", actual=(len(y0),), expected=(self.n_states,))
-        if isinstance(theta, (list, tuple)) and not len(theta) == self.n_theta:
+        if isinstance(theta, list | tuple) and not len(theta) == self.n_theta:
             raise ShapeError(
                 "Length of theta is wrong.", actual=(len(theta),), expected=(self.n_theta,)
             )

pymc/ode/utils.py

@@ -107,7 +107,7 @@ def augment_system(ode_func, n_states, n_theta):
         t_yhat = pt.atleast_1d(yhat)
     else:
         # Stack the results of the ode_func into a single tensor variable
-        if not isinstance(yhat, (list, tuple)):
+        if not isinstance(yhat, list | tuple):
             raise TypeError(
                 f"Unexpected type, {type(yhat)}, returned by ode_func. TensorVariable, list or tuple is expected."
             )