diff --git a/CHANGELOG.md b/CHANGELOG.md
index 4f6ca5c784..8488f7aa5c 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -1,5 +1,13 @@
 # Changelog
 
+## [0.43.0](https://www.github.com/googleapis/gapic-generator-python/compare/v0.42.2...v0.43.0) (2021-03-11)
+
+
+### Features
+
+* add bazel support for gapic metadata ([#811](https://www.github.com/googleapis/gapic-generator-python/issues/811)) ([7ced24a](https://www.github.com/googleapis/gapic-generator-python/commit/7ced24a0b20cb6505587b946c03b1b038eef4b4a))
+* update templates to permit enum aliases ([#809](https://www.github.com/googleapis/gapic-generator-python/issues/809)) ([2e7ea11](https://www.github.com/googleapis/gapic-generator-python/commit/2e7ea11f80210459106f9780e5f013e2a0381d29))
+
 ### [0.42.2](https://www.github.com/googleapis/gapic-generator-python/compare/v0.42.1...v0.42.2) (2021-03-05)
 
 
diff --git a/gapic/ads-templates/%namespace/%name/%version/%sub/types/_enum.py.j2 b/gapic/ads-templates/%namespace/%name/%version/%sub/types/_enum.py.j2
index c9f4cb0c4f..73994a158c 100644
--- a/gapic/ads-templates/%namespace/%name/%version/%sub/types/_enum.py.j2
+++ b/gapic/ads-templates/%namespace/%name/%version/%sub/types/_enum.py.j2
@@ -1,5 +1,8 @@
 class {{ enum.name }}({{ p }}.Enum):
     r"""{{ enum.meta.doc|rst(indent=4) }}"""
+    {% if enum.enum_pb.HasField("options") -%}
+     _pb_options = {{ enum.options_dict }}
+    {% endif -%}
     {% for enum_value in enum.values -%}
     {{ enum_value.name }} = {{ enum_value.number }}
     {% endfor -%}
diff --git a/gapic/ads-templates/setup.py.j2 b/gapic/ads-templates/setup.py.j2
index 18f06803dd..111ce8fb40 100644
--- a/gapic/ads-templates/setup.py.j2
+++ b/gapic/ads-templates/setup.py.j2
@@ -19,7 +19,7 @@ setuptools.setup(
         'google-api-core >= 1.22.2, < 2.0.0dev',
         'googleapis-common-protos >= 1.5.8',
         'grpcio >= 1.10.0',
-        'proto-plus >= 1.4.0',
+        'proto-plus >= 1.15.0',
     {%- if api.requires_package(('google', 'iam', 'v1')) %}
         'grpc-google-iam-v1',
     {%- endif %}
diff --git a/gapic/schema/wrappers.py b/gapic/schema/wrappers.py
index 812630720b..f6ae04ea3e 100644
--- a/gapic/schema/wrappers.py
+++ b/gapic/schema/wrappers.py
@@ -39,6 +39,7 @@
 from google.api import resource_pb2
 from google.api_core import exceptions      # type: ignore
 from google.protobuf import descriptor_pb2  # type: ignore
+from google.protobuf.json_format import MessageToDict  # type: ignore
 
 from gapic import utils
 from gapic.schema import metadata
@@ -413,7 +414,7 @@ def get_field(self, *field_path: str,
         # Get the first field in the path.
         first_field = field_path[0]
         cursor = self.fields[first_field +
-            ('_' if first_field in utils.RESERVED_NAMES else '')]
+                             ('_' if first_field in utils.RESERVED_NAMES else '')]
 
         # Base case: If this is the last field in the path, return it outright.
         if len(field_path) == 1:
@@ -536,6 +537,18 @@ def with_context(self, *, collisions: FrozenSet[str]) -> 'EnumType':
             meta=self.meta.with_context(collisions=collisions),
         )
 
+    @property
+    def options_dict(self) -> Dict:
+        """Return the EnumOptions (if present) as a dict.
+
+        This is a hack to support a pythonic structure representation for
+        the generator templates.
+        """
+        return MessageToDict(
+            self.enum_pb.options,
+            preserving_proto_field_name=True
+        )
+
 
 @dataclasses.dataclass(frozen=True)
 class PythonType:
@@ -869,7 +882,7 @@ def paged_result_field(self) -> Optional[Field]:
 
         # The request must have page_token and next_page_token as they keep track of pages
         for source, source_type, name in ((self.input, str, 'page_token'),
-                           (self.output, str, 'next_page_token')):
+                                          (self.output, str, 'next_page_token')):
             field = source.fields.get(name, None)
             if not field or field.type != source_type:
                 return None
diff --git a/gapic/templates/%namespace/%name_%version/%sub/services/%service/transports/base.py.j2 b/gapic/templates/%namespace/%name_%version/%sub/services/%service/transports/base.py.j2
index 08b5c4b20b..c2db7df705 100644
--- a/gapic/templates/%namespace/%name_%version/%sub/services/%service/transports/base.py.j2
+++ b/gapic/templates/%namespace/%name_%version/%sub/services/%service/transports/base.py.j2
@@ -80,6 +80,9 @@ class {{ service.name }}Transport(abc.ABC):
             host += ':443'
         self._host = host
 
+        # Save the scopes.
+        self._scopes = scopes or self.AUTH_SCOPES
+
         # If no credentials are provided, then determine the appropriate
         # defaults.
         if credentials and credentials_file:
@@ -88,19 +91,16 @@ class {{ service.name }}Transport(abc.ABC):
         if credentials_file is not None:
             credentials, _ = auth.load_credentials_from_file(
                                 credentials_file,
-                                scopes=scopes,
+                                scopes=self._scopes,
                                 quota_project_id=quota_project_id
                             )
 
         elif credentials is None:
-            credentials, _ = auth.default(scopes=scopes, quota_project_id=quota_project_id)
+            credentials, _ = auth.default(scopes=self._scopes, quota_project_id=quota_project_id)
 
         # Save the credentials.
         self._credentials = credentials
 
-        # Lifted into its own function so it can be stubbed out during tests.
-        self._prep_wrapped_messages(client_info)
-
 
     def _prep_wrapped_messages(self, client_info):
         # Precompute the wrapped methods.
diff --git a/gapic/templates/%namespace/%name_%version/%sub/services/%service/transports/grpc.py.j2 b/gapic/templates/%namespace/%name_%version/%sub/services/%service/transports/grpc.py.j2
index b1d1d18917..f03d09f672 100644
--- a/gapic/templates/%namespace/%name_%version/%sub/services/%service/transports/grpc.py.j2
+++ b/gapic/templates/%namespace/%name_%version/%sub/services/%service/transports/grpc.py.j2
@@ -101,7 +101,12 @@ class {{ service.name }}GrpcTransport({{ service.name }}Transport):
           google.api_core.exceptions.DuplicateCredentialArgs: If both ``credentials``
               and ``credentials_file`` are passed.
         """
+        self._grpc_channel = None
         self._ssl_channel_credentials = ssl_channel_credentials
+        self._stubs: Dict[str, Callable] = {}
+        {%- if service.has_lro %}
+        self._operations_client = None
+        {%- endif %}
 
         if api_mtls_endpoint:
             warnings.warn("api_mtls_endpoint is deprecated", DeprecationWarning)
@@ -109,62 +114,50 @@ class {{ service.name }}GrpcTransport({{ service.name }}Transport):
             warnings.warn("client_cert_source is deprecated", DeprecationWarning)
 
         if channel:
-            # Sanity check: Ensure that channel and credentials are not both
-            # provided.
+            # Ignore credentials if a channel was passed.
             credentials = False
-
             # If a channel was explicitly provided, set it.
             self._grpc_channel = channel
             self._ssl_channel_credentials = None
-        elif api_mtls_endpoint:
-            host = api_mtls_endpoint if ":" in api_mtls_endpoint else api_mtls_endpoint + ":443"
-
-            if credentials is None:
-                credentials, _ = auth.default(scopes=self.AUTH_SCOPES, quota_project_id=quota_project_id)
-
-            # Create SSL credentials with client_cert_source or application
-            # default SSL credentials.
-            if client_cert_source:
-                cert, key = client_cert_source()
-                ssl_credentials = grpc.ssl_channel_credentials(
-                    certificate_chain=cert, private_key=key
-                )
-            else:
-                ssl_credentials = SslCredentials().ssl_credentials
-
-            # create a new channel. The provided one is ignored.
-            self._grpc_channel = type(self).create_channel(
-                host,
-                credentials=credentials,
-                credentials_file=credentials_file,
-                ssl_credentials=ssl_credentials,
-                scopes=scopes or self.AUTH_SCOPES,
-                quota_project_id=quota_project_id,
-                options=[
-                    ("grpc.max_send_message_length", -1),
-                    ("grpc.max_receive_message_length", -1),
-                ],
-            )
-            self._ssl_channel_credentials = ssl_credentials
+        
         else:
-            host = host if ":" in host else host + ":443"
-
-            if credentials is None:
-                credentials, _ = auth.default(scopes=self.AUTH_SCOPES, quota_project_id=quota_project_id)
-
-            if client_cert_source_for_mtls and not ssl_channel_credentials:
-                cert, key = client_cert_source_for_mtls()
-                self._ssl_channel_credentials = grpc.ssl_channel_credentials(
-                    certificate_chain=cert, private_key=key
-                )
+            if api_mtls_endpoint:
+                host = api_mtls_endpoint
+
+                # Create SSL credentials with client_cert_source or application
+                # default SSL credentials.
+                if client_cert_source:
+                    cert, key = client_cert_source()
+                    self._ssl_channel_credentials = grpc.ssl_channel_credentials(
+                        certificate_chain=cert, private_key=key
+                    )
+                else:
+                    self._ssl_channel_credentials = SslCredentials().ssl_credentials
             
-            # create a new channel. The provided one is ignored.
+            else:
+                if client_cert_source_for_mtls and not ssl_channel_credentials:
+                    cert, key = client_cert_source_for_mtls()
+                    self._ssl_channel_credentials = grpc.ssl_channel_credentials(
+                        certificate_chain=cert, private_key=key
+                    )
+        
+        # The base transport sets the host, credentials and scopes
+        super().__init__(
+            host=host,
+            credentials=credentials,
+            credentials_file=credentials_file,
+            scopes=scopes,
+            quota_project_id=quota_project_id,
+            client_info=client_info,
+        )
+
+        if not self._grpc_channel:
             self._grpc_channel = type(self).create_channel(
-                host,
-                credentials=credentials,
+                self._host,
+                credentials=self._credentials,
                 credentials_file=credentials_file,
+                scopes=self._scopes,
                 ssl_credentials=self._ssl_channel_credentials,
-                scopes=scopes or self.AUTH_SCOPES,
                 quota_project_id=quota_project_id,
                 options=[
                     ("grpc.max_send_message_length", -1),
@@ -172,20 +165,9 @@ class {{ service.name }}GrpcTransport({{ service.name }}Transport):
                 ],
             )
 
-        self._stubs = {}  # type: Dict[str, Callable]
-        {%- if service.has_lro %}
-        self._operations_client = None
-        {%- endif %}
+        # Wrap messages. This must be done after self._grpc_channel exists
+        self._prep_wrapped_messages(client_info)
 
-        # Run the base constructor.
-        super().__init__(
-            host=host,
-            credentials=credentials,
-            credentials_file=credentials_file,
-            scopes=scopes or self.AUTH_SCOPES,
-            quota_project_id=quota_project_id,
-            client_info=client_info,
-        )
 
     @classmethod
     def create_channel(cls,
@@ -197,7 +179,7 @@ class {{ service.name }}GrpcTransport({{ service.name }}Transport):
                        **kwargs) -> grpc.Channel:
         """Create and return a gRPC channel object.
         Args:
-            address (Optional[str]): The host for the channel to use.
+            host (Optional[str]): The host for the channel to use.
             credentials (Optional[~.Credentials]): The
                 authorization credentials to attach to requests. These
                 credentials identify this application to the service. If
diff --git a/gapic/templates/%namespace/%name_%version/%sub/services/%service/transports/grpc_asyncio.py.j2 b/gapic/templates/%namespace/%name_%version/%sub/services/%service/transports/grpc_asyncio.py.j2
index 94da8db76a..614802d1d7 100644
--- a/gapic/templates/%namespace/%name_%version/%sub/services/%service/transports/grpc_asyncio.py.j2
+++ b/gapic/templates/%namespace/%name_%version/%sub/services/%service/transports/grpc_asyncio.py.j2
@@ -56,7 +56,7 @@ class {{ service.grpc_asyncio_transport_name }}({{ service.name }}Transport):
                        **kwargs) -> aio.Channel:
         """Create and return a gRPC AsyncIO channel object.
         Args:
-            address (Optional[str]): The host for the channel to use.
+            host (Optional[str]): The host for the channel to use.
             credentials (Optional[~.Credentials]): The
                 authorization credentials to attach to requests. These
                 credentials identify this application to the service. If
@@ -145,7 +145,12 @@ class {{ service.grpc_asyncio_transport_name }}({{ service.name }}Transport):
           google.api_core.exceptions.DuplicateCredentialArgs: If both ``credentials``
               and ``credentials_file`` are passed.
         """
+        self._grpc_channel = None
         self._ssl_channel_credentials = ssl_channel_credentials
+        self._stubs: Dict[str, Callable] = {}
+        {%- if service.has_lro %}
+        self._operations_client = None
+        {%- endif %}
     
         if api_mtls_endpoint:
             warnings.warn("api_mtls_endpoint is deprecated", DeprecationWarning)
@@ -153,62 +158,50 @@ class {{ service.grpc_asyncio_transport_name }}({{ service.name }}Transport):
             warnings.warn("client_cert_source is deprecated", DeprecationWarning)
 
         if channel:
-            # Sanity check: Ensure that channel and credentials are not both
-            # provided.
+            # Ignore credentials if a channel was passed.
             credentials = False
-
             # If a channel was explicitly provided, set it.
             self._grpc_channel = channel
             self._ssl_channel_credentials = None
-        elif api_mtls_endpoint:
-            host = api_mtls_endpoint if ":" in api_mtls_endpoint else api_mtls_endpoint + ":443"
-
-            if credentials is None:
-                credentials, _ = auth.default(scopes=self.AUTH_SCOPES, quota_project_id=quota_project_id)
-
-            # Create SSL credentials with client_cert_source or application
-            # default SSL credentials.
-            if client_cert_source:
-                cert, key = client_cert_source()
-                ssl_credentials = grpc.ssl_channel_credentials(
-                    certificate_chain=cert, private_key=key
-                )
-            else:
-                ssl_credentials = SslCredentials().ssl_credentials
 
-            # create a new channel. The provided one is ignored.
-            self._grpc_channel = type(self).create_channel(
-                host,
-                credentials=credentials,
-                credentials_file=credentials_file,
-                ssl_credentials=ssl_credentials,
-                scopes=scopes or self.AUTH_SCOPES,
-                quota_project_id=quota_project_id,
-                options=[
-                    ("grpc.max_send_message_length", -1),
-                    ("grpc.max_receive_message_length", -1),
-                ],
-            )
-            self._ssl_channel_credentials = ssl_credentials
         else:
-            host = host if ":" in host else host + ":443"
+            if api_mtls_endpoint:
+                host = api_mtls_endpoint
+
+                # Create SSL credentials with client_cert_source or application
+                # default SSL credentials.
+                if client_cert_source:
+                    cert, key = client_cert_source()
+                    self._ssl_channel_credentials = grpc.ssl_channel_credentials(
+                        certificate_chain=cert, private_key=key
+                    )
+                else:
+                    self._ssl_channel_credentials = SslCredentials().ssl_credentials
+            
+            else:
+                if client_cert_source_for_mtls and not ssl_channel_credentials:
+                    cert, key = client_cert_source_for_mtls()
+                    self._ssl_channel_credentials = grpc.ssl_channel_credentials(
+                        certificate_chain=cert, private_key=key
+                    )
 
-            if credentials is None:
-                credentials, _ = auth.default(scopes=self.AUTH_SCOPES, quota_project_id=quota_project_id)
+        # The base transport sets the host, credentials and scopes
+        super().__init__(
+            host=host,
+            credentials=credentials,
+            credentials_file=credentials_file,
+            scopes=scopes,
+            quota_project_id=quota_project_id,
+            client_info=client_info,
+        )
 
-            if client_cert_source_for_mtls and not ssl_channel_credentials:
-                cert, key = client_cert_source_for_mtls()
-                self._ssl_channel_credentials = grpc.ssl_channel_credentials(
-                    certificate_chain=cert, private_key=key
-                )
-            
-            # create a new channel. The provided one is ignored.
+        if not self._grpc_channel:
             self._grpc_channel = type(self).create_channel(
-                host,
-                credentials=credentials,
+                self._host,
+                credentials=self._credentials,
                 credentials_file=credentials_file,
+                scopes=self._scopes,
                 ssl_credentials=self._ssl_channel_credentials,
-                scopes=scopes or self.AUTH_SCOPES,
                 quota_project_id=quota_project_id,
                 options=[
                     ("grpc.max_send_message_length", -1),
@@ -216,20 +209,8 @@ class {{ service.grpc_asyncio_transport_name }}({{ service.name }}Transport):
                 ],
             )
 
-        # Run the base constructor.
-        super().__init__(
-            host=host,
-            credentials=credentials,
-            credentials_file=credentials_file,
-            scopes=scopes or self.AUTH_SCOPES,
-            quota_project_id=quota_project_id,
-            client_info=client_info,
-        )
-
-        self._stubs = {}
-        {%- if service.has_lro %}
-        self._operations_client = None
-        {%- endif %}
+        # Wrap messages. This must be done after self._grpc_channel exists
+        self._prep_wrapped_messages(client_info)
 
     @property
     def grpc_channel(self) -> aio.Channel:
diff --git a/gapic/templates/%namespace/%name_%version/%sub/services/%service/transports/rest.py.j2 b/gapic/templates/%namespace/%name_%version/%sub/services/%service/transports/rest.py.j2
index bb6acbadd6..f368273d16 100644
--- a/gapic/templates/%namespace/%name_%version/%sub/services/%service/transports/rest.py.j2
+++ b/gapic/templates/%namespace/%name_%version/%sub/services/%service/transports/rest.py.j2
@@ -92,6 +92,7 @@ class {{ service.name }}RestTransport({{ service.name }}Transport):
         {%- endif %}
         if client_cert_source_for_mtls:
             self._session.configure_mtls_channel(client_cert_source_for_mtls)
+        self._prep_wrapped_messages(client_info)
 
     {%- if service.has_lro %}
     
diff --git a/gapic/templates/%namespace/%name_%version/%sub/types/_enum.py.j2 b/gapic/templates/%namespace/%name_%version/%sub/types/_enum.py.j2
index c9f4cb0c4f..73994a158c 100644
--- a/gapic/templates/%namespace/%name_%version/%sub/types/_enum.py.j2
+++ b/gapic/templates/%namespace/%name_%version/%sub/types/_enum.py.j2
@@ -1,5 +1,8 @@
 class {{ enum.name }}({{ p }}.Enum):
     r"""{{ enum.meta.doc|rst(indent=4) }}"""
+    {% if enum.enum_pb.HasField("options") -%}
+     _pb_options = {{ enum.options_dict }}
+    {% endif -%}
     {% for enum_value in enum.values -%}
     {{ enum_value.name }} = {{ enum_value.number }}
     {% endfor -%}
diff --git a/gapic/templates/setup.py.j2 b/gapic/templates/setup.py.j2
index 94af4ae760..f7ed0a9923 100644
--- a/gapic/templates/setup.py.j2
+++ b/gapic/templates/setup.py.j2
@@ -28,7 +28,7 @@ setuptools.setup(
     install_requires=(
         'google-api-core[grpc] >= 1.22.2, < 2.0.0dev',
         'libcst >= 0.2.5',
-        'proto-plus >= 1.4.0',
+        'proto-plus >= 1.15.0',
     {%- if api.requires_package(('google', 'iam', 'v1')) or opts.add_iam_methods %}
         'grpc-google-iam-v1',
     {%- endif %}
diff --git a/requirements.txt b/requirements.txt
index 17d86ccef8..2770b54084 100644
--- a/requirements.txt
+++ b/requirements.txt
@@ -3,7 +3,7 @@ google-api-core==1.26.1
 googleapis-common-protos==1.53.0
 jinja2==2.11.3
 MarkupSafe==1.1.1
-protobuf==3.15.5
+protobuf==3.15.6
 pypandoc==1.5
 PyYAML==5.4.1
 dataclasses==0.6  # TODO(busunkim) remove when 3.6 support is dropped
diff --git a/rules_python_gapic/py_gapic.bzl b/rules_python_gapic/py_gapic.bzl
index 7c6f938f52..bcb55fdb93 100644
--- a/rules_python_gapic/py_gapic.bzl
+++ b/rules_python_gapic/py_gapic.bzl
@@ -18,13 +18,20 @@ def py_gapic_library(
         name,
         srcs,
         grpc_service_config = None,
-        plugin_args = [],
-        opt_args = [],
+        plugin_args = None,
+        opt_args = None,
+        metadata = False,
         **kwargs):
     #    srcjar_target_name = "%s_srcjar" % name
     srcjar_target_name = name
     srcjar_output_suffix = ".srcjar"
 
+    plugin_args = plugin_args or []
+    opt_args = opt_args or []
+
+    if metadata:
+        plugin_args.append("metadata")
+
     file_args = {}
     if grpc_service_config:
         file_args[grpc_service_config] =  "retry-config"
diff --git a/test_utils/test_utils.py b/test_utils/test_utils.py
index beab26518f..2aafab454a 100644
--- a/test_utils/test_utils.py
+++ b/test_utils/test_utils.py
@@ -290,6 +290,7 @@ def make_enum(
     module: str = 'baz',
     values: typing.Tuple[str, int] = (),
     meta: metadata.Metadata = None,
+    options: desc.EnumOptions = None,
 ) -> wrappers.EnumType:
     enum_value_pbs = [
         desc.EnumValueDescriptorProto(name=i[0], number=i[1])
@@ -298,6 +299,7 @@ def make_enum(
     enum_pb = desc.EnumDescriptorProto(
         name=name,
         value=enum_value_pbs,
+        options=options,
     )
     return wrappers.EnumType(
         enum_pb=enum_pb,
diff --git a/tests/unit/schema/wrappers/test_enums.py b/tests/unit/schema/wrappers/test_enums.py
index 3debb5603b..2eeb9c043f 100644
--- a/tests/unit/schema/wrappers/test_enums.py
+++ b/tests/unit/schema/wrappers/test_enums.py
@@ -37,6 +37,18 @@ def test_enum_value_properties():
 
 
 def test_enum_ident():
-    message = make_enum('Baz', package='foo.v1', module='bar')
-    assert str(message.ident) == 'bar.Baz'
-    assert message.ident.sphinx == 'foo.v1.bar.Baz'
+    enum = make_enum('Baz', package='foo.v1', module='bar')
+    assert str(enum.ident) == 'bar.Baz'
+    assert enum.ident.sphinx == 'foo.v1.bar.Baz'
+
+
+def test_enum_options_dict():
+    cephalopod = make_enum("Cephalopod", package="animalia.v1",
+                     module="mollusca", options={"allow_alias": True})
+    assert isinstance(cephalopod.enum_pb.options, descriptor_pb2.EnumOptions)
+    assert cephalopod.options_dict == {"allow_alias": True}
+
+    bivalve = make_enum("Bivalve", package="animalia.v1",
+                     module="mollusca")
+    assert isinstance(bivalve.enum_pb.options, descriptor_pb2.EnumOptions)
+    assert bivalve.options_dict == {}