[refactor] remove @solid decorator (#10952)

### Summary & Motivation

Remove the `@solid` decorator and update all references of it to `@op`.
This also involved updating `input_defs` and `output_defs` to `ins` and
`out` respectively.

Internal companion PR: dagster-io/internal#4900

One GQL snapshot (for `dynamic_pipeline` around line ~950 of
`graphql/repo.py`) updated as a result of switch to `DynamicOut`.

### How I Tested These Changes

BK

examples/docs_snippets/docs_snippets/concepts/repositories_workspaces/my_working_directory/hello_world_repository.py

@@ -1,8 +1,8 @@
-from dagster import repository
-from dagster._legacy import pipeline, solid
+from dagster import op, repository
+from dagster._legacy import pipeline
 
 
-@solid
+@op
 def hello_world():
     pass
 

examples/docs_snippets/docs_snippets/intro_tutorial/airflow.py

@@ -2,10 +2,11 @@
 
 import requests
 
-from dagster._legacy import pipeline, solid
+from dagster import op
+from dagster._legacy import pipeline
 
 
-@solid
+@op
 def hello_cereal(context):
     response = requests.get("https://docs.dagster.io/assets/cereal.csv")
     lines = response.text.split("\n")

examples/docs_snippets/docs_snippets_tests/concepts_tests/assets_tests/test_materialization_io_managers.py

@@ -1,7 +1,7 @@
 import pandas as pd
 
-from dagster import io_manager
-from dagster._legacy import ModeDefinition, execute_pipeline, pipeline, solid
+from dagster import io_manager, op
+from dagster._legacy import ModeDefinition, execute_pipeline, pipeline
 from docs_snippets.concepts.assets.materialization_io_managers import (
     PandasCsvIOManager,
     PandasCsvIOManagerWithAsset,
@@ -18,7 +18,7 @@ def _generate_pipeline_for_io_manager(manager, config_schema=None):
     def custom_io_manager(_):
         return manager
 
-    @solid
+    @op
     def dummy_solid():
         return DummyClass.from_dict({"some_column": [2]})
 

examples/docs_snippets/docs_snippets_tests/intro_tutorial_tests/test_type_guide.py

@@ -11,21 +11,17 @@
     AssetMaterialization,
     DagsterType,
     DagsterTypeCheckDidNotPass,
+    In,
     Nothing,
+    Out,
     PythonObjectDagsterType,
     dagster_type_loader,
     dagster_type_materializer,
     make_python_type_usable_as_dagster_type,
+    op,
     usable_as_dagster_type,
 )
-from dagster._legacy import (
-    InputDefinition,
-    OutputDefinition,
-    execute_pipeline,
-    execute_solid,
-    pipeline,
-    solid,
-)
+from dagster._legacy import execute_pipeline, execute_solid, pipeline
 from dagster._utils import safe_tempfile_path
 
 
@@ -38,7 +34,7 @@ def test_basic_even_type():
     # end_test_basic_even_type
 
     # start_test_basic_even_type_with_annotations
-    @solid
+    @op
     def double_even(num: EvenDagsterType) -> EvenDagsterType:
         # These type annotations are a shorthand for constructing InputDefinitions
         # and OutputDefinitions, and are not mypy compliant
@@ -63,9 +59,9 @@ def test_basic_even_type_no_annotations():
     )
 
     # start_test_basic_even_type_no_annotations
-    @solid(
-        input_defs=[InputDefinition("num", EvenDagsterType)],
-        output_defs=[OutputDefinition(EvenDagsterType)],
+    @op(
+        ins={"num": In(EvenDagsterType)},
+        out=Out(EvenDagsterType),
     )
     def double_even(num):
         return num
@@ -93,7 +89,7 @@ def __init__(self, num):
     # end_object_type
 
     # start_use_object_type
-    @solid
+    @op
     def double_even(even_num: EvenDagsterType) -> EvenDagsterType:
         # These type annotations are a shorthand for constructing InputDefinitions
         # and OutputDefinitions, and are not mypy compliant
@@ -120,13 +116,13 @@ def load_even_type(_, cfg):
     EvenDagsterType = PythonObjectDagsterType(EvenType, loader=load_even_type)
     # end_type_loader
 
-    @solid
+    @op
     def double_even(even_num: EvenDagsterType) -> EvenDagsterType:
         return EvenType(even_num.num * 2)
 
     # start_via_config
     yaml_doc = """
-    solids:
+    ops:
         double_even:
             inputs:
                 even_num: 2
@@ -135,7 +131,7 @@ def double_even(even_num: EvenDagsterType) -> EvenDagsterType:
     assert execute_solid(double_even, run_config=yaml.safe_load(yaml_doc)).success
 
     assert execute_solid(
-        double_even, run_config={"solids": {"double_even": {"inputs": {"even_num": 2}}}}
+        double_even, run_config={"ops": {"double_even": {"inputs": {"even_num": 2}}}}
     ).success
 
     # Same same as above w/r/t chatting to prha
@@ -166,24 +162,24 @@ def save_to_file_materialization(_, cfg, value):
         EvenType, materializer=save_to_file_materialization
     )
 
-    @solid
+    @op
     def double_even(even_num: EvenDagsterType) -> EvenDagsterType:
         return EvenType(even_num.num * 2)
 
     with safe_tempfile_path() as path:
         yaml_doc = """
-solids:
+ops:
     double_even:
         outputs:
             - result:
                 path: {path}
  """
-        solid_result = execute_solid(
+        op_result = execute_solid(
             double_even,
             input_values={"even_num": EvenType(2)},
             run_config=yaml.safe_load(yaml_doc.format(path=path)),
         )
-        assert solid_result.success
+        assert op_result.success
 
 
 def test_mypy_compliance():
@@ -198,7 +194,7 @@ def __init__(self, num):
     else:
         EvenDagsterType = PythonObjectDagsterType(EvenType)
 
-    @solid
+    @op
     def double_even(even_num: EvenDagsterType) -> EvenDagsterType:
         return EvenType(even_num.num * 2)
 
@@ -207,11 +203,11 @@ def double_even(even_num: EvenDagsterType) -> EvenDagsterType:
 
 
 def test_nothing_type():
-    @solid(output_defs=[OutputDefinition(Nothing, "cleanup_done")])
+    @op(out={"cleanup_done": Out(Nothing)})
     def do_cleanup():
         pass
 
-    @solid(input_defs=[InputDefinition("on_cleanup_done", Nothing)])
+    @op(ins={"on_cleanup_done": In(Nothing)})
     def after_cleanup():  # Argument not required for Nothing types
         return "worked"
 
@@ -227,23 +223,23 @@ def nothing_pipeline():
 def test_nothing_fanin_actually_test():
     ordering = {"counter": 0}
 
-    @solid(output_defs=[OutputDefinition(Nothing)])
+    @op(out=Out(Nothing))
     def start_first_pipeline_section(context):
         ordering["counter"] += 1
-        ordering[context.solid.name] = ordering["counter"]
+        ordering[context.op.name] = ordering["counter"]
 
-    @solid(
-        input_defs=[InputDefinition("first_section_done", Nothing)],
-        output_defs=[OutputDefinition(dagster_type=Nothing)],
+    @op(
+        ins={"first_section_done": In(Nothing)},
+        out=Out(Nothing),
     )
     def perform_clean_up(context):
         ordering["counter"] += 1
-        ordering[context.solid.name] = ordering["counter"]
+        ordering[context.op.name] = ordering["counter"]
 
-    @solid(input_defs=[InputDefinition("on_cleanup_tasks_done", Nothing)])
+    @op(ins={"on_cleanup_tasks_done": In(Nothing)})
     def start_next_pipeline_section(context):
         ordering["counter"] += 1
-        ordering[context.solid.name] = ordering["counter"]
+        ordering[context.op.name] = ordering["counter"]
         return "worked"
 
     @pipeline
@@ -264,18 +260,20 @@ def fanin_pipeline():
 
 
 def test_nothing_fanin_empty_body_for_guide():
-    @solid(output_defs=[OutputDefinition(Nothing)])
+    @op(out=Out(Nothing))
     def start_first_pipeline_section():
         pass
 
-    @solid(
-        input_defs=[InputDefinition("first_section_done", Nothing)],
-        output_defs=[OutputDefinition(dagster_type=Nothing)],
+    @op(
+        ins={"first_section_done": In(Nothing)},
+        out=Out(Nothing),
     )
     def perform_clean_up():
         pass
 
-    @solid(input_defs=[InputDefinition("on_cleanup_tasks_done", Nothing)])
+    @op(
+        ins={"on_cleanup_tasks_done": In(Nothing)},
+    )
     def start_next_pipeline_section():
         pass
 
@@ -302,7 +300,7 @@ def __init__(self, num):
             self.num = num
 
     # end_usable_as
-    @solid
+    @op
     def double_even(even_num: EvenType) -> EvenType:
         return EvenType(even_num.num * 2)
 
@@ -323,7 +321,7 @@ def __init__(self, num):
 
     make_python_type_usable_as_dagster_type(EvenType, EvenDagsterType)
 
-    @solid
+    @op
     def double_even(even_num: EvenType) -> EvenType:
         return EvenType(even_num.num * 2)
 

integration_tests/test_suites/daemon-test-suite/repo.py

@@ -1,8 +1,8 @@
-from dagster import RunRequest, repository, schedule, sensor
-from dagster._legacy import pipeline, solid
+from dagster import RunRequest, op, repository, schedule, sensor
+from dagster._legacy import pipeline
 
 
-@solid()
+@op()
 def foo_solid(_):
     pass
 

integration_tests/test_suites/daemon-test-suite/test_memory.py

@@ -4,14 +4,14 @@
 from contextlib import contextmanager
 
 import objgraph
-from dagster import RunRequest, repository, schedule, sensor
+from dagster import RunRequest, op, repository, schedule, sensor
 from dagster._core.test_utils import instance_for_test
 from dagster._core.workspace.load_target import PythonFileTarget
 from dagster._daemon.controller import daemon_controller_from_instance
-from dagster._legacy import pipeline, solid
+from dagster._legacy import pipeline
 
 
-@solid()
+@op()
 def foo_solid(_):
     pass
 

python_modules/automation/automation/docker/images/k8s-example/build_cache/example_project/example_repo/repo.py

@@ -1,25 +1,24 @@
 import os
 from collections import Counter
 
-from dagster import file_relative_path, repository
+from dagster import In, file_relative_path, repository
+from dagster._core.definitions.decorators import op
 from dagster._legacy import (
-    InputDefinition,
     ModeDefinition,
     PresetDefinition,
     default_executors,
     pipeline,
-    solid,
 )
 from dagster_aws.s3 import s3_pickle_io_manager, s3_resource
 from dagster_celery_k8s import celery_k8s_job_executor
 
 
-@solid(input_defs=[InputDefinition("word", str)], config_schema={"factor": int})
+@op(ins={"word": In()}, config_schema={"factor": int})
 def multiply_the_word(context, word):
     return word * context.op_config["factor"]
 
 
-@solid(input_defs=[InputDefinition("word")])
+@op(ins={"word": In()})
 def count_letters(_context, word):
     return dict(Counter(word))
 

python_modules/dagit/dagit_tests/override_pipeline.py

@@ -1,13 +1,14 @@
-from dagster import Int, repository
-from dagster._legacy import InputDefinition, OutputDefinition, pipeline, solid
+from dagster import In, Out, repository
+from dagster._core.definitions.decorators import op
+from dagster._legacy import pipeline
 
 
-@solid(input_defs=[InputDefinition("num", Int)], output_defs=[OutputDefinition(Int)])
+@op(ins={"num": In(int)}, out=Out(int))
 def add_one(num):
     return num + 1
 
 
-@solid(input_defs=[InputDefinition("num", Int)], output_defs=[OutputDefinition(Int)])
+@op(ins={"num": In(int)}, out=Out(int))
 def mult_two(num):
     return num * 2
 

python_modules/dagit/dagit_tests/pipeline.py

@@ -1,14 +1,15 @@
-from dagster import Int, repository
+from dagster import In, Out, repository
+from dagster._core.definitions.decorators import op
 from dagster._core.test_utils import today_at_midnight
-from dagster._legacy import InputDefinition, OutputDefinition, daily_schedule, pipeline, solid
+from dagster._legacy import daily_schedule, pipeline
 
 
-@solid(input_defs=[InputDefinition("num", Int)], output_defs=[OutputDefinition(Int)])
+@op(ins={"num": In(int)}, out=Out(int))
 def add_one(num):
     return num + 1
 
 
-@solid(input_defs=[InputDefinition("num", Int)], output_defs=[OutputDefinition(Int)])
+@op(ins={"num": In(int)}, out=Out(int))
 def mult_two(num):
     return num * 2
 

python_modules/dagit/dagit_tests/test_debug_cli.py

@@ -4,11 +4,12 @@
 from click.testing import CliRunner
 from dagit.debug import dagit_debug_command
 from dagster._cli.debug import export_command
+from dagster._core.definitions.decorators import op
 from dagster._core.test_utils import instance_for_test
-from dagster._legacy import execute_pipeline, pipeline, solid
+from dagster._legacy import execute_pipeline, pipeline
 
 
-@solid
+@op
 def emit_one():
     return 1
 

python_modules/dagit/dagit_tests/test_subscriptions.py

@@ -7,10 +7,11 @@
 import pytest
 from dagit.graphql import GraphQLWS
 from dagit.webserver import DagitWebserver
+from dagster._core.definitions.decorators import op
 from dagster._core.test_utils import environ, instance_for_test
 from dagster._core.workspace.context import WorkspaceProcessContext
 from dagster._core.workspace.load_target import WorkspaceFileTarget
-from dagster._legacy import execute_pipeline, pipeline, solid
+from dagster._legacy import execute_pipeline, pipeline
 from dagster._utils import file_relative_path
 from starlette.testclient import TestClient
 
@@ -72,7 +73,7 @@ def end_subscription(ws):
     ws.close()
 
 
-@solid
+@op
 def example_solid():
     return 1