class_docstr: Class that contains methods for loading an Ax experiment from SQLAlchemy.
  
  Instantiate with an instance of Config to customize the functionality.
  For even more flexibility, create a subclass.
  
  Attributes:
      config: Metadata needed to save and load an experiment to SQLAlchemy.
class_docstr_tok: ["Class","that","contains","methods","for","loading","an","Ax","experiment","from","SQLAlchemy",".","Instantiate","with","an","instance","of","Config","to","customize","the","functionality",".","For","even","more","flexibility",",","create","a","subclass",".","Attributes",":","config",":","Metadata","needed","to","save","and","load","an","experiment","to","SQLAlchemy","."]
docstr: Convert SQLAlchemy Trial to Ax Trial.
  
  Args:
      trial_sqa: `SQATrial` to decode.
      reduced_state: Whether to load trial's generator run(s) with a slightly
      reduced state (without model state, search space, and optimization config).
docstr_tok: ["Convert","SQLAlchemy","Trial","to","Ax","Trial",".","Args",":","trial_sqa",":","`","SQATrial","`","to","decode",".","reduced_state",":","Whether","to","load","trial","'s","generator","run","(","s",")","with","a","slightly","reduced","state","(","without","model","state",",","search","space",",","and","optimization","config",")","."]
code: def trial_from_sqa(
      self,
      trial_sqa: SQATrial,
      experiment: Experiment,
      reduced_state: bool = False,
      ax_object_field_overrides: Optional[Dict[str, Any]] = None,
  ) -> BaseTrial:
      """Convert SQLAlchemy Trial to Ax Trial.
  
      Args:
          trial_sqa: `SQATrial` to decode.
          reduced_state: Whether to load trial's generator run(s) with a slightly
          reduced state (without model state, search space, and optimization config).
  
      """
      immutable_ss_and_oc = (
          experiment.immutable_search_space_and_opt_config
      )
      if trial_sqa.is_batch:
          trial = BatchTrial(
              experiment=experiment,
              optimize_for_power=trial_sqa.optimize_for_power,
              ttl_seconds=trial_sqa.ttl_seconds,
              index=trial_sqa.index,
              lifecycle_stage=trial_sqa.lifecycle_stage,
          )
          generator_run_structs = [
              GeneratorRunStruct(
                  generator_run=self.generator_run_from_sqa(
                      generator_run_sqa=generator_run_sqa,
                      reduced_state=reduced_state,
                      immutable_search_space_and_opt_config=immutable_ss_and_oc,
                  ),
                  weight=generator_run_sqa.weight or 1.0,
              )
              for generator_run_sqa in trial_sqa.generator_runs
          ]
          if trial_sqa.status_quo_name is not None:
              new_generator_run_structs = []
              for struct in generator_run_structs:
                  if (
                      struct.generator_run.generator_run_type
                      == GeneratorRunType.STATUS_QUO.name
                  ):
                      status_quo_weight = struct.generator_run.weights[
                          0
                      ]
                      trial._status_quo = struct.generator_run.arms[0]
                      trial._status_quo_weight_override = (
                          status_quo_weight
                      )
                      trial._status_quo_generator_run_db_id = (
                          struct.generator_run.db_id
                      )
                      trial._status_quo_arm_db_id = (
                          struct.generator_run.arms[0].db_id
                      )
                  else:
                      new_generator_run_structs.append(struct)
              generator_run_structs = new_generator_run_structs
          trial._generator_run_structs = generator_run_structs
          if not reduced_state:
              trial._abandoned_arms_metadata = {
                  abandoned_arm_sqa.name: self.abandoned_arm_from_sqa(
                      abandoned_arm_sqa=abandoned_arm_sqa
                  )
                  for abandoned_arm_sqa in trial_sqa.abandoned_arms
              }
          trial._refresh_arms_by_name()  # Trigger cache build
      else:
          trial = Trial(
              experiment=experiment,
              ttl_seconds=trial_sqa.ttl_seconds,
              index=trial_sqa.index,
          )
          if trial_sqa.generator_runs:
              if len(trial_sqa.generator_runs) != 1:
                  raise SQADecodeError(
                      "Cannot decode SQATrial to Trial because trial is not batched "
                      "but has more than one generator run."
                  )
              trial._generator_run = self.generator_run_from_sqa(
                  generator_run_sqa=trial_sqa.generator_runs[0],
                  reduced_state=reduced_state,
                  immutable_search_space_and_opt_config=immutable_ss_and_oc,
              )
      trial._trial_type = trial_sqa.trial_type
      # Swap `DISPATCHED` for `RUNNING`, since `DISPATCHED` is deprecated and nearly
      # equivalent to `RUNNING`.
      trial._status = (
          trial_sqa.status
          if trial_sqa.status != TrialStatus.DISPATCHED
          else TrialStatus.RUNNING
      )
      trial._time_created = trial_sqa.time_created
      trial._time_completed = trial_sqa.time_completed
      trial._time_staged = trial_sqa.time_staged
      trial._time_run_started = trial_sqa.time_run_started
      trial._abandoned_reason = trial_sqa.abandoned_reason
      trial._failed_reason = trial_sqa.failed_reason
      # pyre-fixme[9]: _run_metadata has type `Dict[str, Any]`; used as
      #  `Optional[Dict[str, Any]]`.
      # pyre-fixme[8]: Attribute has type `Dict[str, typing.Any]`; used as
      #  `Optional[typing.Dict[Variable[_KT], Variable[_VT]]]`.
      trial._run_metadata = (
          dict(trial_sqa.run_metadata)
          if trial_sqa.run_metadata is not None
          else None
      )
      # pyre-fixme[9]: _run_metadata has type `Dict[str, Any]`; used as
      #  `Optional[Dict[str, Any]]`.
      # pyre-fixme[8]: Attribute has type `Dict[str, typing.Any]`; used as
      #  `Optional[typing.Dict[Variable[_KT], Variable[_VT]]]`.
      trial._stop_metadata = (
          dict(trial_sqa.stop_metadata)
          if trial_sqa.stop_metadata is not None
          else None
      )
      trial._num_arms_created = trial_sqa.num_arms_created
      trial._runner = (
          self.runner_from_sqa(
              trial_sqa.runner,
              runner_kwargs=(
                  ax_object_field_overrides.get("runner")
                  if ax_object_field_overrides is not None
                  else None
              ),
          )
          if trial_sqa.runner
          else None
      )
      trial._generation_step_index = trial_sqa.generation_step_index
      trial._properties = dict(trial_sqa.properties or {})
      trial.db_id = trial_sqa.id
      return trial
  
imports: import json
  from collections import defaultdict, OrderedDict
  from enum import Enum
  from logging import Logger
  from typing import Any, cast, Dict, List, Optional, Tuple, Type, Union
  import pandas
  from ax.core.arm import Arm
  from ax.core.base_trial import BaseTrial, TrialStatus
  from ax.core.batch_trial import AbandonedArm, BatchTrial, GeneratorRunStruct
  from ax.core.data import Data
  from ax.core.experiment import Experiment
  from ax.core.generator_run import GeneratorRun, GeneratorRunType
  from ax.core.metric import Metric
  from ax.core.multi_type_experiment import MultiTypeExperiment
  from ax.core.objective import MultiObjective, Objective, ScalarizedObjective
  from ax.core.optimization_config import MultiObjectiveOptimizationConfig, OptimizationConfig
  from ax.core.outcome_constraint import ObjectiveThreshold, OutcomeConstraint, ScalarizedOutcomeConstraint
  from ax.core.parameter import ChoiceParameter, FixedParameter, Parameter, RangeParameter
  from ax.core.parameter_constraint import OrderConstraint, ParameterConstraint, SumConstraint
  from ax.core.parameter_distribution import ParameterDistribution
  from ax.core.risk_measures import RiskMeasure
  from ax.core.runner import Runner
  from ax.core.search_space import HierarchicalSearchSpace, RobustSearchSpace, SearchSpace
  from ax.core.trial import Trial
  from ax.exceptions.storage import SQADecodeError
  from ax.modelbridge.generation_strategy import GenerationStrategy
  from ax.storage.json_store.decoder import object_from_json
  from ax.storage.sqa_store.db import session_scope
  from ax.storage.sqa_store.sqa_classes import SQAAbandonedArm, SQAArm, SQAData, SQAExperiment, SQAGenerationStrategy, SQAGeneratorRun, SQAMetric, SQAParameter, SQAParameterConstraint, SQARunner, SQATrial
  from ax.storage.sqa_store.sqa_config import SQAConfig
  from ax.storage.utils import DomainType, MetricIntent, ParameterConstraintType
  from ax.utils.common.constants import Keys
  from ax.utils.common.logger import get_logger
  from ax.utils.common.typeutils import not_none
  from sqlalchemy.orm.exc import DetachedInstanceError
label_desc: Use image node_id 18 for calling the Decoder obj's underlying member method code with example usage: obj.trial_from_sqa(trial_sqa, experiment, reduced_state, ax_object_field_overrides) and returns: trial