diff --git a/sdk/python/feast/base_feature_view.py b/sdk/python/feast/base_feature_view.py
index b2178ec631..b0cd70617f 100644
--- a/sdk/python/feast/base_feature_view.py
+++ b/sdk/python/feast/base_feature_view.py
@@ -33,32 +33,12 @@ class BaseFeatureView(ABC):
 
     @abstractmethod
     def __init__(self, name: str, features: List[Feature]):
-        self._name = name
-        self._features = features
-        self._projection = FeatureViewProjection.from_definition(self)
+        self.name = name
+        self.features = features
+        self.projection = FeatureViewProjection.from_definition(self)
         self.created_timestamp: Optional[datetime] = None
         self.last_updated_timestamp: Optional[datetime] = None
 
-    @property
-    def name(self) -> str:
-        return self._name
-
-    @property
-    def features(self) -> List[Feature]:
-        return self._features
-
-    @features.setter
-    def features(self, value):
-        self._features = value
-
-    @property
-    def projection(self) -> FeatureViewProjection:
-        return self._projection
-
-    @projection.setter
-    def projection(self, value):
-        self._projection = value
-
     @property
     @abstractmethod
     def proto_class(self) -> Type[Message]:
diff --git a/sdk/python/feast/data_source.py b/sdk/python/feast/data_source.py
index 94910c6c08..360b9b8542 100644
--- a/sdk/python/feast/data_source.py
+++ b/sdk/python/feast/data_source.py
@@ -44,51 +44,9 @@ class KafkaOptions:
     def __init__(
         self, bootstrap_servers: str, message_format: StreamFormat, topic: str,
     ):
-        self._bootstrap_servers = bootstrap_servers
-        self._message_format = message_format
-        self._topic = topic
-
-    @property
-    def bootstrap_servers(self):
-        """
-        Returns a comma-separated list of Kafka bootstrap servers
-        """
-        return self._bootstrap_servers
-
-    @bootstrap_servers.setter
-    def bootstrap_servers(self, bootstrap_servers):
-        """
-        Sets a comma-separated list of Kafka bootstrap servers
-        """
-        self._bootstrap_servers = bootstrap_servers
-
-    @property
-    def message_format(self):
-        """
-        Returns the data format that is used to encode the feature data in Kafka messages
-        """
-        return self._message_format
-
-    @message_format.setter
-    def message_format(self, message_format):
-        """
-        Sets the data format that is used to encode the feature data in Kafka messages
-        """
-        self._message_format = message_format
-
-    @property
-    def topic(self):
-        """
-        Returns the Kafka topic to collect feature data from
-        """
-        return self._topic
-
-    @topic.setter
-    def topic(self, topic):
-        """
-        Sets the Kafka topic to collect feature data from
-        """
-        self._topic = topic
+        self.bootstrap_servers = bootstrap_servers
+        self.message_format = message_format
+        self.topic = topic
 
     @classmethod
     def from_proto(cls, kafka_options_proto: DataSourceProto.KafkaOptions):
@@ -135,51 +93,9 @@ class KinesisOptions:
     def __init__(
         self, record_format: StreamFormat, region: str, stream_name: str,
     ):
-        self._record_format = record_format
-        self._region = region
-        self._stream_name = stream_name
-
-    @property
-    def record_format(self):
-        """
-        Returns the data format used to encode the feature data in the Kinesis records.
-        """
-        return self._record_format
-
-    @record_format.setter
-    def record_format(self, record_format):
-        """
-        Sets the data format used to encode the feature data in the Kinesis records.
-        """
-        self._record_format = record_format
-
-    @property
-    def region(self):
-        """
-        Returns the AWS region of Kinesis stream
-        """
-        return self._region
-
-    @region.setter
-    def region(self, region):
-        """
-        Sets the AWS region of Kinesis stream
-        """
-        self._region = region
-
-    @property
-    def stream_name(self):
-        """
-        Returns the Kinesis stream name to obtain feature data from
-        """
-        return self._stream_name
-
-    @stream_name.setter
-    def stream_name(self, stream_name):
-        """
-        Sets the Kinesis stream name to obtain feature data from
-        """
-        self._stream_name = stream_name
+        self.record_format = record_format
+        self.region = region
+        self.stream_name = stream_name
 
     @classmethod
     def from_proto(cls, kinesis_options_proto: DataSourceProto.KinesisOptions):
@@ -233,10 +149,10 @@ class DataSource(ABC):
         date_partition_column (optional): Timestamp column used for partitioning.
     """
 
-    _event_timestamp_column: str
-    _created_timestamp_column: str
-    _field_mapping: Dict[str, str]
-    _date_partition_column: str
+    event_timestamp_column: str
+    created_timestamp_column: str
+    field_mapping: Dict[str, str]
+    date_partition_column: str
 
     def __init__(
         self,
@@ -246,14 +162,14 @@ def __init__(
         date_partition_column: Optional[str] = None,
     ):
         """Creates a DataSource object."""
-        self._event_timestamp_column = (
+        self.event_timestamp_column = (
             event_timestamp_column if event_timestamp_column else ""
         )
-        self._created_timestamp_column = (
+        self.created_timestamp_column = (
             created_timestamp_column if created_timestamp_column else ""
         )
-        self._field_mapping = field_mapping if field_mapping else {}
-        self._date_partition_column = (
+        self.field_mapping = field_mapping if field_mapping else {}
+        self.date_partition_column = (
             date_partition_column if date_partition_column else ""
         )
 
@@ -271,62 +187,6 @@ def __eq__(self, other):
 
         return True
 
-    @property
-    def field_mapping(self) -> Dict[str, str]:
-        """
-        Returns the field mapping of this data source.
-        """
-        return self._field_mapping
-
-    @field_mapping.setter
-    def field_mapping(self, field_mapping):
-        """
-        Sets the field mapping of this data source.
-        """
-        self._field_mapping = field_mapping
-
-    @property
-    def event_timestamp_column(self) -> str:
-        """
-        Returns the event timestamp column of this data source.
-        """
-        return self._event_timestamp_column
-
-    @event_timestamp_column.setter
-    def event_timestamp_column(self, event_timestamp_column):
-        """
-        Sets the event timestamp column of this data source.
-        """
-        self._event_timestamp_column = event_timestamp_column
-
-    @property
-    def created_timestamp_column(self) -> str:
-        """
-        Returns the created timestamp column of this data source.
-        """
-        return self._created_timestamp_column
-
-    @created_timestamp_column.setter
-    def created_timestamp_column(self, created_timestamp_column):
-        """
-        Sets the created timestamp column of this data source.
-        """
-        self._created_timestamp_column = created_timestamp_column
-
-    @property
-    def date_partition_column(self) -> str:
-        """
-        Returns the date partition column of this data source.
-        """
-        return self._date_partition_column
-
-    @date_partition_column.setter
-    def date_partition_column(self, date_partition_column):
-        """
-        Sets the date partition column of this data source.
-        """
-        self._date_partition_column = date_partition_column
-
     @staticmethod
     @abstractmethod
     def from_proto(data_source: DataSourceProto) -> Any:
@@ -450,7 +310,7 @@ def __init__(
             field_mapping,
             date_partition_column,
         )
-        self._kafka_options = KafkaOptions(
+        self.kafka_options = KafkaOptions(
             bootstrap_servers=bootstrap_servers,
             message_format=message_format,
             topic=topic,
@@ -472,20 +332,6 @@ def __eq__(self, other):
 
         return True
 
-    @property
-    def kafka_options(self):
-        """
-        Returns the kafka options of this data source
-        """
-        return self._kafka_options
-
-    @kafka_options.setter
-    def kafka_options(self, kafka_options):
-        """
-        Sets the kafka options of this data source
-        """
-        self._kafka_options = kafka_options
-
     @staticmethod
     def from_proto(data_source: DataSourceProto):
         return KafkaSource(
@@ -531,30 +377,16 @@ class RequestDataSource(DataSource):
     def source_datatype_to_feast_value_type() -> Callable[[str], ValueType]:
         raise NotImplementedError
 
-    _name: str
-    _schema: Dict[str, ValueType]
+    name: str
+    schema: Dict[str, ValueType]
 
     def __init__(
         self, name: str, schema: Dict[str, ValueType],
     ):
         """Creates a RequestDataSource object."""
         super().__init__()
-        self._name = name
-        self._schema = schema
-
-    @property
-    def name(self) -> str:
-        """
-        Returns the name of this data source
-        """
-        return self._name
-
-    @property
-    def schema(self) -> Dict[str, ValueType]:
-        """
-        Returns the schema for this request data source
-        """
-        return self._schema
+        self.name = name
+        self.schema = schema
 
     def validate(self, config: RepoConfig):
         pass
@@ -576,9 +408,9 @@ def from_proto(data_source: DataSourceProto):
 
     def to_proto(self) -> DataSourceProto:
         schema_pb = {}
-        for key, value in self._schema.items():
+        for key, value in self.schema.items():
             schema_pb[key] = value.value
-        options = DataSourceProto.RequestDataOptions(name=self._name, schema=schema_pb)
+        options = DataSourceProto.RequestDataOptions(name=self.name, schema=schema_pb)
         data_source_proto = DataSourceProto(
             type=DataSourceProto.REQUEST_SOURCE, request_data_options=options
         )
@@ -629,7 +461,7 @@ def __init__(
             field_mapping,
             date_partition_column,
         )
-        self._kinesis_options = KinesisOptions(
+        self.kinesis_options = KinesisOptions(
             record_format=record_format, region=region, stream_name=stream_name
         )
 
@@ -651,20 +483,6 @@ def __eq__(self, other):
 
         return True
 
-    @property
-    def kinesis_options(self):
-        """
-        Returns the kinesis options of this data source
-        """
-        return self._kinesis_options
-
-    @kinesis_options.setter
-    def kinesis_options(self, kinesis_options):
-        """
-        Sets the kinesis options of this data source
-        """
-        self._kinesis_options = kinesis_options
-
     def to_proto(self) -> DataSourceProto:
         data_source_proto = DataSourceProto(
             type=DataSourceProto.STREAM_KINESIS,
diff --git a/sdk/python/feast/entity.py b/sdk/python/feast/entity.py
index 85045b392c..efac8c17da 100644
--- a/sdk/python/feast/entity.py
+++ b/sdk/python/feast/entity.py
@@ -45,14 +45,14 @@ class Entity:
         last_updated_timestamp: The time when the entity was last updated.
     """
 
-    _name: str
-    _value_type: ValueType
-    _join_key: str
-    _description: str
-    _tags: Dict[str, str]
-    _owner: str
-    _created_timestamp: Optional[datetime]
-    _last_updated_timestamp: Optional[datetime]
+    name: str
+    value_type: ValueType
+    join_key: str
+    description: str
+    tags: Dict[str, str]
+    owner: str
+    created_timestamp: Optional[datetime]
+    last_updated_timestamp: Optional[datetime]
 
     @log_exceptions
     def __init__(
@@ -66,13 +66,13 @@ def __init__(
         owner: str = "",
     ):
         """Creates an Entity object."""
-        self._name = name
-        self._value_type = value_type
-        self._join_key = join_key if join_key else name
-        self._description = description
+        self.name = name
+        self.value_type = value_type
+        self.join_key = join_key if join_key else name
+        self.description = description
 
         if labels is not None:
-            self._tags = labels
+            self.tags = labels
             warnings.warn(
                 (
                     "The parameter 'labels' is being deprecated. Please use 'tags' instead. "
@@ -81,11 +81,11 @@ def __init__(
                 DeprecationWarning,
             )
         else:
-            self._tags = labels or tags or {}
+            self.tags = labels or tags or {}
 
-        self._owner = owner
-        self._created_timestamp = None
-        self._last_updated_timestamp = None
+        self.owner = owner
+        self.created_timestamp = None
+        self.last_updated_timestamp = None
 
     def __hash__(self) -> int:
         return hash((id(self), self.name))
@@ -109,78 +109,6 @@ def __eq__(self, other):
     def __str__(self):
         return str(MessageToJson(self.to_proto()))
 
-    @property
-    def name(self) -> str:
-        return self._name
-
-    @name.setter
-    def name(self, name: str):
-        self._name = name
-
-    @property
-    def value_type(self) -> ValueType:
-        return self._value_type
-
-    @value_type.setter
-    def value_type(self, value_type: ValueType):
-        self._value_type = value_type
-
-    @property
-    def join_key(self) -> str:
-        return self._join_key
-
-    @join_key.setter
-    def join_key(self, join_key: str):
-        self._join_key = join_key
-
-    @property
-    def description(self) -> str:
-        return self._description
-
-    @description.setter
-    def description(self, description: str):
-        self._description = description
-
-    @property
-    def tags(self) -> Dict[str, str]:
-        return self._tags
-
-    @tags.setter
-    def tags(self, tags: Dict[str, str]):
-        self._tags = tags
-
-    @property
-    def labels(self) -> Dict[str, str]:
-        return self._tags
-
-    @labels.setter
-    def labels(self, tags: Dict[str, str]):
-        self._tags = tags
-
-    @property
-    def owner(self) -> str:
-        return self._owner
-
-    @owner.setter
-    def owner(self, owner: str):
-        self._owner = owner
-
-    @property
-    def created_timestamp(self) -> Optional[datetime]:
-        return self._created_timestamp
-
-    @created_timestamp.setter
-    def created_timestamp(self, created_timestamp: datetime):
-        self._created_timestamp = created_timestamp
-
-    @property
-    def last_updated_timestamp(self) -> Optional[datetime]:
-        return self._last_updated_timestamp
-
-    @last_updated_timestamp.setter
-    def last_updated_timestamp(self, last_updated_timestamp: datetime):
-        self._last_updated_timestamp = last_updated_timestamp
-
     def is_valid(self):
         """
         Validates the state of this entity locally.
diff --git a/sdk/python/feast/feature_service.py b/sdk/python/feast/feature_service.py
index bb6ec909bf..40030b34ce 100644
--- a/sdk/python/feast/feature_service.py
+++ b/sdk/python/feast/feature_service.py
@@ -36,13 +36,13 @@ class FeatureService:
         last_updated_timestamp: The time when the feature service was last updated.
     """
 
-    _name: str
-    _feature_view_projections: List[FeatureViewProjection]
-    _description: str
-    _tags: Dict[str, str]
-    _owner: str
-    _created_timestamp: Optional[datetime] = None
-    _last_updated_timestamp: Optional[datetime] = None
+    name: str
+    feature_view_projections: List[FeatureViewProjection]
+    description: str
+    tags: Dict[str, str]
+    owner: str
+    created_timestamp: Optional[datetime] = None
+    last_updated_timestamp: Optional[datetime] = None
 
     @log_exceptions
     def __init__(
@@ -59,23 +59,23 @@ def __init__(
         Raises:
             ValueError: If one of the specified features is not a valid type.
         """
-        self._name = name
-        self._feature_view_projections = []
+        self.name = name
+        self.feature_view_projections = []
 
         for feature_grouping in features:
             if isinstance(feature_grouping, BaseFeatureView):
-                self._feature_view_projections.append(feature_grouping.projection)
+                self.feature_view_projections.append(feature_grouping.projection)
             else:
                 raise ValueError(
                     f"The feature service {name} has been provided with an invalid type "
                     f'{type(feature_grouping)} as part of the "features" argument.)'
                 )
 
-        self._description = description
-        self._tags = tags or {}
-        self._owner = owner
-        self._created_timestamp = None
-        self._last_updated_timestamp = None
+        self.description = description
+        self.tags = tags or {}
+        self.owner = owner
+        self.created_timestamp = None
+        self.last_updated_timestamp = None
 
     def __repr__(self):
         items = (f"{k} = {v}" for k, v in self.__dict__.items())
@@ -108,64 +108,6 @@ def __eq__(self, other):
 
         return True
 
-    @property
-    def name(self) -> str:
-        return self._name
-
-    @name.setter
-    def name(self, name: str):
-        self._name = name
-
-    @property
-    def feature_view_projections(self) -> List[FeatureViewProjection]:
-        return self._feature_view_projections
-
-    @feature_view_projections.setter
-    def feature_view_projections(
-        self, feature_view_projections: List[FeatureViewProjection]
-    ):
-        self._feature_view_projections = feature_view_projections
-
-    @property
-    def description(self) -> str:
-        return self._description
-
-    @description.setter
-    def description(self, description: str):
-        self._description = description
-
-    @property
-    def tags(self) -> Dict[str, str]:
-        return self._tags
-
-    @tags.setter
-    def tags(self, tags: Dict[str, str]):
-        self._tags = tags
-
-    @property
-    def owner(self) -> str:
-        return self._owner
-
-    @owner.setter
-    def owner(self, owner: str):
-        self._owner = owner
-
-    @property
-    def created_timestamp(self) -> Optional[datetime]:
-        return self._created_timestamp
-
-    @created_timestamp.setter
-    def created_timestamp(self, created_timestamp: datetime):
-        self._created_timestamp = created_timestamp
-
-    @property
-    def last_updated_timestamp(self) -> Optional[datetime]:
-        return self._last_updated_timestamp
-
-    @last_updated_timestamp.setter
-    def last_updated_timestamp(self, last_updated_timestamp: datetime):
-        self._last_updated_timestamp = last_updated_timestamp
-
     @classmethod
     def from_proto(cls, feature_service_proto: FeatureServiceProto):
         """
diff --git a/sdk/python/feast/infra/offline_stores/bigquery_source.py b/sdk/python/feast/infra/offline_stores/bigquery_source.py
index f97f687b0f..4c4d2a591c 100644
--- a/sdk/python/feast/infra/offline_stores/bigquery_source.py
+++ b/sdk/python/feast/infra/offline_stores/bigquery_source.py
@@ -22,7 +22,7 @@ def __init__(
         date_partition_column: Optional[str] = "",
         query: Optional[str] = None,
     ):
-        self._bigquery_options = BigQueryOptions(table_ref=table_ref, query=query)
+        self.bigquery_options = BigQueryOptions(table_ref=table_ref, query=query)
 
         super().__init__(
             event_timestamp_column,
@@ -47,25 +47,11 @@ def __eq__(self, other):
 
     @property
     def table_ref(self):
-        return self._bigquery_options.table_ref
+        return self.bigquery_options.table_ref
 
     @property
     def query(self):
-        return self._bigquery_options.query
-
-    @property
-    def bigquery_options(self):
-        """
-        Returns the bigquery options of this data source
-        """
-        return self._bigquery_options
-
-    @bigquery_options.setter
-    def bigquery_options(self, bigquery_options):
-        """
-        Sets the bigquery options of this data source
-        """
-        self._bigquery_options = bigquery_options
+        return self.bigquery_options.query
 
     @staticmethod
     def from_proto(data_source: DataSourceProto):
diff --git a/sdk/python/feast/infra/offline_stores/file_source.py b/sdk/python/feast/infra/offline_stores/file_source.py
index 7d52110985..1c1431043c 100644
--- a/sdk/python/feast/infra/offline_stores/file_source.py
+++ b/sdk/python/feast/infra/offline_stores/file_source.py
@@ -59,7 +59,7 @@ def __init__(
         else:
             file_url = path
 
-        self._file_options = FileOptions(
+        self.file_options = FileOptions(
             file_format=file_format,
             file_url=file_url,
             s3_endpoint_override=s3_endpoint_override,
@@ -86,26 +86,12 @@ def __eq__(self, other):
             == other.file_options.s3_endpoint_override
         )
 
-    @property
-    def file_options(self):
-        """
-        Returns the file options of this data source
-        """
-        return self._file_options
-
-    @file_options.setter
-    def file_options(self, file_options):
-        """
-        Sets the file options of this data source
-        """
-        self._file_options = file_options
-
     @property
     def path(self):
         """
-        Returns the file path of this feature data source
+        Returns the path of this file data source.
         """
-        return self._file_options.file_url
+        return self.file_options.file_url
 
     @staticmethod
     def from_proto(data_source: DataSourceProto):
@@ -144,7 +130,7 @@ def get_table_column_names_and_types(
         self, config: RepoConfig
     ) -> Iterable[Tuple[str, str]]:
         filesystem, path = FileSource.create_filesystem_and_path(
-            self.path, self._file_options.s3_endpoint_override
+            self.path, self.file_options.s3_endpoint_override
         )
         schema = ParquetFile(
             path if filesystem is None else filesystem.open_input_file(path)
diff --git a/sdk/python/feast/infra/offline_stores/redshift_source.py b/sdk/python/feast/infra/offline_stores/redshift_source.py
index 949f1c9221..398374b6a8 100644
--- a/sdk/python/feast/infra/offline_stores/redshift_source.py
+++ b/sdk/python/feast/infra/offline_stores/redshift_source.py
@@ -48,7 +48,7 @@ def __init__(
         # The default Redshift schema is named "public".
         _schema = "public" if table and not schema else schema
 
-        self._redshift_options = RedshiftOptions(
+        self.redshift_options = RedshiftOptions(
             table=table, schema=_schema, query=query
         )
 
@@ -91,27 +91,17 @@ def __eq__(self, other):
     @property
     def table(self):
         """Returns the table of this Redshift source."""
-        return self._redshift_options.table
+        return self.redshift_options.table
 
     @property
     def schema(self):
         """Returns the schema of this Redshift source."""
-        return self._redshift_options.schema
+        return self.redshift_options.schema
 
     @property
     def query(self):
         """Returns the Redshift options of this Redshift source."""
-        return self._redshift_options.query
-
-    @property
-    def redshift_options(self):
-        """Returns the Redshift options of this Redshift source."""
-        return self._redshift_options
-
-    @redshift_options.setter
-    def redshift_options(self, _redshift_options):
-        """Sets the Redshift options of this Redshift source."""
-        self._redshift_options = _redshift_options
+        return self.redshift_options.query
 
     def to_proto(self) -> DataSourceProto:
         """
diff --git a/sdk/python/feast/infra/offline_stores/snowflake_source.py b/sdk/python/feast/infra/offline_stores/snowflake_source.py
index b5d50be0f4..4e1553f486 100644
--- a/sdk/python/feast/infra/offline_stores/snowflake_source.py
+++ b/sdk/python/feast/infra/offline_stores/snowflake_source.py
@@ -50,7 +50,7 @@ def __init__(
         # The default Snowflake schema is named "PUBLIC".
         _schema = "PUBLIC" if (database and table and not schema) else schema
 
-        self._snowflake_options = SnowflakeOptions(
+        self.snowflake_options = SnowflakeOptions(
             database=database, schema=_schema, table=table, query=query
         )
 
@@ -95,32 +95,22 @@ def __eq__(self, other):
     @property
     def database(self):
         """Returns the database of this snowflake source."""
-        return self._snowflake_options.database
+        return self.snowflake_options.database
 
     @property
     def schema(self):
         """Returns the schema of this snowflake source."""
-        return self._snowflake_options.schema
+        return self.snowflake_options.schema
 
     @property
     def table(self):
         """Returns the table of this snowflake source."""
-        return self._snowflake_options.table
+        return self.snowflake_options.table
 
     @property
     def query(self):
         """Returns the snowflake options of this snowflake source."""
-        return self._snowflake_options.query
-
-    @property
-    def snowflake_options(self):
-        """Returns the snowflake options of this snowflake source."""
-        return self._snowflake_options
-
-    @snowflake_options.setter
-    def snowflake_options(self, _snowflake_options):
-        """Sets the snowflake options of this snowflake source."""
-        self._snowflake_options = _snowflake_options
+        return self.snowflake_options.query
 
     def to_proto(self) -> DataSourceProto:
         """
diff --git a/sdk/python/feast/registry.py b/sdk/python/feast/registry.py
index 4273493255..1c47274067 100644
--- a/sdk/python/feast/registry.py
+++ b/sdk/python/feast/registry.py
@@ -235,8 +235,8 @@ def apply_entity(self, entity: Entity, project: str, commit: bool = True):
 
         now = datetime.utcnow()
         if not entity.created_timestamp:
-            entity._created_timestamp = now
-        entity._last_updated_timestamp = now
+            entity.created_timestamp = now
+        entity.last_updated_timestamp = now
 
         entity_proto = entity.to_proto()
         entity_proto.spec.project = project