diff --git a/backend/api_v2/__init__.py b/backend/api_v2/__init__.py
new file mode 100644
index 0000000000..e69de29bb2
diff --git a/backend/api_v2/admin.py b/backend/api_v2/admin.py
new file mode 100644
index 0000000000..37f0837a74
--- /dev/null
+++ b/backend/api_v2/admin.py
@@ -0,0 +1,5 @@
+from django.contrib import admin
+
+from .models import APIDeployment, APIKey
+
+admin.site.register([APIDeployment, APIKey])
diff --git a/backend/api_v2/api_deployment_views.py b/backend/api_v2/api_deployment_views.py
new file mode 100644
index 0000000000..802d8f8bc6
--- /dev/null
+++ b/backend/api_v2/api_deployment_views.py
@@ -0,0 +1,152 @@
+import json
+import logging
+from typing import Any, Optional
+
+from api_v2.constants import ApiExecution
+from api_v2.deployment_helper import DeploymentHelper
+from api_v2.exceptions import InvalidAPIRequest, NoActiveAPIKeyError
+from api_v2.models import APIDeployment
+from api_v2.postman_collection.dto import PostmanCollection
+from api_v2.serializers import (
+    APIDeploymentListSerializer,
+    APIDeploymentSerializer,
+    DeploymentResponseSerializer,
+    ExecutionRequestSerializer,
+)
+from django.db.models import QuerySet
+from django.http import HttpResponse
+from permissions.permission import IsOwner
+from rest_framework import serializers, status, views, viewsets
+from rest_framework.decorators import action
+from rest_framework.request import Request
+from rest_framework.response import Response
+from rest_framework.serializers import Serializer
+from utils.enums import CeleryTaskState
+from workflow_manager.workflow_v2.dto import ExecutionResponse
+
+logger = logging.getLogger(__name__)
+
+
+class DeploymentExecution(views.APIView):
+    def initialize_request(
+        self, request: Request, *args: Any, **kwargs: Any
+    ) -> Request:
+        """To remove csrf request for public API.
+
+        Args:
+            request (Request): _description_
+
+        Returns:
+            Request: _description_
+        """
+        setattr(request, "csrf_processing_done", True)
+        return super().initialize_request(request, *args, **kwargs)
+
+    @DeploymentHelper.validate_api_key
+    def post(
+        self, request: Request, org_name: str, api_name: str, api: APIDeployment
+    ) -> Response:
+        file_objs = request.FILES.getlist(ApiExecution.FILES_FORM_DATA)
+        serializer = ExecutionRequestSerializer(data=request.data)
+        serializer.is_valid(raise_exception=True)
+        timeout = serializer.get_timeout(serializer.validated_data)
+        include_metadata = (
+            request.data.get(ApiExecution.INCLUDE_METADATA, "false").lower() == "true"
+        )
+        if not file_objs or len(file_objs) == 0:
+            raise InvalidAPIRequest("File shouldn't be empty")
+        response = DeploymentHelper.execute_workflow(
+            organization_name=org_name,
+            api=api,
+            file_objs=file_objs,
+            timeout=timeout,
+            include_metadata=include_metadata,
+        )
+        if "error" in response and response["error"]:
+            return Response(
+                {"message": response},
+                status=status.HTTP_422_UNPROCESSABLE_ENTITY,
+            )
+        return Response({"message": response}, status=status.HTTP_200_OK)
+
+    @DeploymentHelper.validate_api_key
+    def get(
+        self, request: Request, org_name: str, api_name: str, api: APIDeployment
+    ) -> Response:
+        execution_id = request.query_params.get("execution_id")
+        if not execution_id:
+            raise InvalidAPIRequest("execution_id shouldn't be empty")
+        response: ExecutionResponse = DeploymentHelper.get_execution_status(
+            execution_id=execution_id
+        )
+        if response.execution_status != CeleryTaskState.SUCCESS.value:
+            return Response(
+                {
+                    "status": response.execution_status,
+                    "message": response.result,
+                },
+                status=status.HTTP_422_UNPROCESSABLE_ENTITY,
+            )
+        return Response(
+            {"status": response.execution_status, "message": response.result},
+            status=status.HTTP_200_OK,
+        )
+
+
+class APIDeploymentViewSet(viewsets.ModelViewSet):
+    permission_classes = [IsOwner]
+
+    def get_queryset(self) -> Optional[QuerySet]:
+        return APIDeployment.objects.filter(created_by=self.request.user)
+
+    def get_serializer_class(self) -> serializers.Serializer:
+        if self.action in ["list"]:
+            return APIDeploymentListSerializer
+        return APIDeploymentSerializer
+
+    @action(detail=True, methods=["get"])
+    def fetch_one(self, request: Request, pk: Optional[str] = None) -> Response:
+        """Custom action to fetch a single instance."""
+        instance = self.get_object()
+        serializer = self.get_serializer(instance)
+        return Response(serializer.data)
+
+    def create(
+        self, request: Request, *args: tuple[Any], **kwargs: dict[str, Any]
+    ) -> Response:
+        serializer: Serializer = self.get_serializer(data=request.data)
+        serializer.is_valid(raise_exception=True)
+        self.perform_create(serializer)
+        api_key = DeploymentHelper.create_api_key(serializer=serializer)
+        response_serializer = DeploymentResponseSerializer(
+            {"api_key": api_key.api_key, **serializer.data}
+        )
+
+        headers = self.get_success_headers(serializer.data)
+        return Response(
+            response_serializer.data,
+            status=status.HTTP_201_CREATED,
+            headers=headers,
+        )
+
+    @action(detail=True, methods=["get"])
+    def download_postman_collection(
+        self, request: Request, pk: Optional[str] = None
+    ) -> Response:
+        """Downloads a Postman Collection of the API deployment instance."""
+        instance = self.get_object()
+        api_key_inst = instance.apikey_set.filter(is_active=True).first()
+        if not api_key_inst:
+            logger.error(f"No active API key set for deployment {instance.pk}")
+            raise NoActiveAPIKeyError(deployment_name=instance.display_name)
+
+        postman_collection = PostmanCollection.create(
+            instance=instance, api_key=api_key_inst.api_key
+        )
+        response = HttpResponse(
+            json.dumps(postman_collection.to_dict()), content_type="application/json"
+        )
+        response["Content-Disposition"] = (
+            f'attachment; filename="{instance.display_name}.json"'
+        )
+        return response
diff --git a/backend/api_v2/api_key_views.py b/backend/api_v2/api_key_views.py
new file mode 100644
index 0000000000..906127c404
--- /dev/null
+++ b/backend/api_v2/api_key_views.py
@@ -0,0 +1,28 @@
+from api_v2.deployment_helper import DeploymentHelper
+from api_v2.exceptions import APINotFound
+from api_v2.key_helper import KeyHelper
+from api_v2.models import APIKey
+from api_v2.serializers import APIKeyListSerializer, APIKeySerializer
+from rest_framework import serializers, viewsets
+from rest_framework.decorators import action
+from rest_framework.request import Request
+from rest_framework.response import Response
+
+
+class APIKeyViewSet(viewsets.ModelViewSet):
+    queryset = APIKey.objects.all()
+
+    def get_serializer_class(self) -> serializers.Serializer:
+        if self.action in ["api_keys"]:
+            return APIKeyListSerializer
+        return APIKeySerializer
+
+    @action(detail=True, methods=["get"])
+    def api_keys(self, request: Request, api_id: str) -> Response:
+        """Custom action to fetch api keys of an api deployment."""
+        api = DeploymentHelper.get_api_by_id(api_id=api_id)
+        if not api:
+            raise APINotFound()
+        keys = KeyHelper.list_api_keys_of_api(api_instance=api)
+        serializer = self.get_serializer(keys, many=True)
+        return Response(serializer.data)
diff --git a/backend/api_v2/apps.py b/backend/api_v2/apps.py
new file mode 100644
index 0000000000..16cbd4ad5b
--- /dev/null
+++ b/backend/api_v2/apps.py
@@ -0,0 +1,5 @@
+from django.apps import AppConfig
+
+
+class ApiConfig(AppConfig):
+    name = "api_v2"
diff --git a/backend/api_v2/constants.py b/backend/api_v2/constants.py
new file mode 100644
index 0000000000..0ec324cc93
--- /dev/null
+++ b/backend/api_v2/constants.py
@@ -0,0 +1,6 @@
+class ApiExecution:
+    PATH: str = "deployment/api"
+    MAXIMUM_TIMEOUT_IN_SEC: int = 300  # 5 minutes
+    FILES_FORM_DATA: str = "files"
+    TIMEOUT_FORM_DATA: str = "timeout"
+    INCLUDE_METADATA: str = "include_metadata"
diff --git a/backend/api_v2/deployment_helper.py b/backend/api_v2/deployment_helper.py
new file mode 100644
index 0000000000..e94c84e465
--- /dev/null
+++ b/backend/api_v2/deployment_helper.py
@@ -0,0 +1,244 @@
+import logging
+import uuid
+from functools import wraps
+from typing import Any, Optional
+from urllib.parse import urlencode
+
+from api_v2.constants import ApiExecution
+from api_v2.exceptions import (
+    ApiKeyCreateException,
+    APINotFound,
+    Forbidden,
+    InactiveAPI,
+    UnauthorizedKey,
+)
+from api_v2.key_helper import KeyHelper
+from api_v2.models import APIDeployment, APIKey
+from api_v2.serializers import APIExecutionResponseSerializer
+from django.core.files.uploadedfile import UploadedFile
+from django.db import connection
+from rest_framework import status
+from rest_framework.request import Request
+from rest_framework.response import Response
+from rest_framework.serializers import Serializer
+from rest_framework.utils.serializer_helpers import ReturnDict
+from utils.constants import Account
+from utils.local_context import StateStore
+from workflow_manager.endpoint_v2.destination import DestinationConnector
+from workflow_manager.endpoint_v2.source import SourceConnector
+from workflow_manager.workflow_v2.dto import ExecutionResponse
+from workflow_manager.workflow_v2.models.workflow import Workflow
+from workflow_manager.workflow_v2.workflow_helper import WorkflowHelper
+
+logger = logging.getLogger(__name__)
+
+
+class DeploymentHelper:
+    @staticmethod
+    def validate_api_key(func: Any) -> Any:
+        """Decorator that validates the API key.
+
+        Sample header:
+            Authorization: Bearer 123e4567-e89b-12d3-a456-426614174001
+        Args:
+            func (Any): Function to wrap for validation
+        """
+
+        @wraps(func)
+        def wrapper(self: Any, request: Request, *args: Any, **kwargs: Any) -> Any:
+            """Wrapper to validate the inputs and key.
+
+            Args:
+                request (Request): Request context
+
+            Raises:
+                Forbidden: _description_
+                APINotFound: _description_
+
+            Returns:
+                Any: _description_
+            """
+            try:
+                authorization_header = request.headers.get("Authorization")
+                api_key = None
+                if authorization_header and authorization_header.startswith("Bearer "):
+                    api_key = authorization_header.split(" ")[1]
+                if not api_key:
+                    raise Forbidden("Missing api key")
+                org_name = kwargs.get("org_name") or request.data.get("org_name")
+                api_name = kwargs.get("api_name") or request.data.get("api_name")
+                if not api_name:
+                    raise Forbidden("Missing api_name")
+                # Set organization in state store for API
+                StateStore.set(Account.ORGANIZATION_ID, org_name)
+
+                api_deployment = DeploymentHelper.get_deployment_by_api_name(
+                    api_name=api_name
+                )
+                DeploymentHelper.validate_api(
+                    api_deployment=api_deployment, api_key=api_key
+                )
+                kwargs["api"] = api_deployment
+                return func(self, request, *args, **kwargs)
+
+            except (UnauthorizedKey, InactiveAPI, APINotFound):
+                raise
+            except Exception as exception:
+                logger.error(f"Exception: {exception}")
+                return Response(
+                    {"error": str(exception)}, status=status.HTTP_403_FORBIDDEN
+                )
+
+        return wrapper
+
+    @staticmethod
+    def validate_api(api_deployment: Optional[APIDeployment], api_key: str) -> None:
+        """Validating API and API key.
+
+        Args:
+            api_deployment (Optional[APIDeployment]): _description_
+            api_key (str): _description_
+
+        Raises:
+            APINotFound: _description_
+            InactiveAPI: _description_
+        """
+        if not api_deployment:
+            raise APINotFound()
+        if not api_deployment.is_active:
+            raise InactiveAPI()
+        KeyHelper.validate_api_key(api_key=api_key, api_instance=api_deployment)
+
+    @staticmethod
+    def validate_and_get_workflow(workflow_id: str) -> Workflow:
+        """Validate that the specified workflow_id exists in the Workflow
+        model."""
+        return WorkflowHelper.get_workflow_by_id(workflow_id)
+
+    @staticmethod
+    def get_api_by_id(api_id: str) -> Optional[APIDeployment]:
+        try:
+            api_deployment: APIDeployment = APIDeployment.objects.get(pk=api_id)
+            return api_deployment
+        except APIDeployment.DoesNotExist:
+            return None
+
+    @staticmethod
+    def construct_complete_endpoint(api_name: str) -> str:
+        """Constructs the complete API endpoint by appending organization
+        schema, endpoint path, and Django app backend URL.
+
+        Parameters:
+        - endpoint (str): The endpoint path to be appended to the complete URL.
+
+        Returns:
+        - str: The complete API endpoint URL.
+        """
+        org_schema = connection.get_tenant().schema_name
+        return f"{ApiExecution.PATH}/{org_schema}/{api_name}/"
+
+    @staticmethod
+    def construct_status_endpoint(api_endpoint: str, execution_id: str) -> str:
+        """Construct a complete status endpoint URL by appending the
+        execution_id as a query parameter.
+
+        Args:
+            api_endpoint (str): The base API endpoint.
+            execution_id (str): The execution ID to be included as
+                a query parameter.
+
+        Returns:
+            str: The complete status endpoint URL.
+        """
+        query_parameters = urlencode({"execution_id": execution_id})
+        complete_endpoint = f"/{api_endpoint}?{query_parameters}"
+        return complete_endpoint
+
+    @staticmethod
+    def get_deployment_by_api_name(
+        api_name: str,
+    ) -> Optional[APIDeployment]:
+        """Get and return the APIDeployment object by api_name."""
+        try:
+            api: APIDeployment = APIDeployment.objects.get(api_name=api_name)
+            return api
+        except APIDeployment.DoesNotExist:
+            return None
+
+    @staticmethod
+    def create_api_key(serializer: Serializer) -> APIKey:
+        """To make API key for an API.
+
+        Args:
+            serializer (Serializer): Request serializer
+
+        Raises:
+            ApiKeyCreateException: Exception
+        """
+        api_deployment: APIDeployment = serializer.instance
+        try:
+            api_key: APIKey = KeyHelper.create_api_key(api_deployment)
+            return api_key
+        except Exception as error:
+            logger.error(f"Error while creating API key error: {str(error)}")
+            api_deployment.delete()
+            logger.info("Deleted the deployment instance")
+            raise ApiKeyCreateException()
+
+    @staticmethod
+    def execute_workflow(
+        organization_name: str,
+        api: APIDeployment,
+        file_objs: list[UploadedFile],
+        timeout: int,
+    ) -> ReturnDict:
+        """Execute workflow by api.
+
+        Args:
+            organization_name (str): organization name
+            api (APIDeployment): api model object
+            file_obj (UploadedFile): input file
+
+        Returns:
+            ReturnDict: execution status/ result
+        """
+        workflow_id = api.workflow.id
+        pipeline_id = api.id
+        execution_id = str(uuid.uuid4())
+        hash_values_of_files = SourceConnector.add_input_file_to_api_storage(
+            workflow_id=workflow_id,
+            execution_id=execution_id,
+            file_objs=file_objs,
+        )
+        try:
+            result = WorkflowHelper.execute_workflow_async(
+                workflow_id=workflow_id,
+                pipeline_id=pipeline_id,
+                hash_values_of_files=hash_values_of_files,
+                timeout=timeout,
+                execution_id=execution_id,
+            )
+            result.status_api = DeploymentHelper.construct_status_endpoint(
+                api_endpoint=api.api_endpoint, execution_id=execution_id
+            )
+        except Exception:
+            DestinationConnector.delete_api_storage_dir(
+                workflow_id=workflow_id, execution_id=execution_id
+            )
+            raise
+        return APIExecutionResponseSerializer(result).data
+
+    @staticmethod
+    def get_execution_status(execution_id: str) -> ExecutionResponse:
+        """Current status of api execution.
+
+        Args:
+            execution_id (str): execution id
+
+        Returns:
+            ReturnDict: status/result of execution
+        """
+        execution_response: ExecutionResponse = WorkflowHelper.get_status_of_async_task(
+            execution_id=execution_id
+        )
+        return execution_response
diff --git a/backend/api_v2/exceptions.py b/backend/api_v2/exceptions.py
new file mode 100644
index 0000000000..c3a58ff054
--- /dev/null
+++ b/backend/api_v2/exceptions.py
@@ -0,0 +1,55 @@
+from typing import Optional
+
+from rest_framework.exceptions import APIException
+
+
+class MandatoryWorkflowId(APIException):
+    status_code = 400
+    default_detail = "Workflow ID is mandatory"
+
+
+class ApiKeyCreateException(APIException):
+    status_code = 500
+    default_detail = "Exception while create API key"
+
+
+class Forbidden(APIException):
+    status_code = 403
+    default_detail = (
+        "User is forbidden from performing this action. Please contact admin"
+    )
+
+
+class APINotFound(APIException):
+    status_code = 404
+    default_detail = "API not found"
+
+
+class InvalidAPIRequest(APIException):
+    status_code = 400
+    default_detail = "Bad request"
+
+
+class InactiveAPI(APIException):
+    status_code = 404
+    default_detail = "API not found or Inactive"
+
+
+class UnauthorizedKey(APIException):
+    status_code = 401
+    default_detail = "Unauthorized"
+
+
+class NoActiveAPIKeyError(APIException):
+    status_code = 409
+    default_detail = "No active API keys configured for this deployment"
+
+    def __init__(
+        self,
+        detail: Optional[str] = None,
+        code: Optional[str] = None,
+        deployment_name: str = "this deployment",
+    ):
+        if detail is None:
+            detail = f"No active API keys configured for {deployment_name}"
+        super().__init__(detail, code)
diff --git a/backend/api_v2/key_helper.py b/backend/api_v2/key_helper.py
new file mode 100644
index 0000000000..2a1a430a9b
--- /dev/null
+++ b/backend/api_v2/key_helper.py
@@ -0,0 +1,72 @@
+import logging
+
+from api_v2.exceptions import Forbidden, UnauthorizedKey
+from api_v2.models import APIDeployment, APIKey
+from api_v2.serializers import APIKeySerializer
+from workflow_manager.workflow_v2.workflow_helper import WorkflowHelper
+
+logger = logging.getLogger(__name__)
+
+
+class KeyHelper:
+    @staticmethod
+    def validate_api_key(api_key: str, api_instance: APIDeployment) -> None:
+        """Validate api key.
+
+        Args:
+            api_key (str): api key from request
+            api_instance (APIDeployment): api deployment instance
+
+        Raises:
+            Forbidden: _description_
+        """
+        try:
+            api_key_instance: APIKey = APIKey.objects.get(api_key=api_key)
+            if not KeyHelper.has_access(api_key_instance, api_instance):
+                raise UnauthorizedKey()
+        except APIKey.DoesNotExist:
+            raise UnauthorizedKey()
+        except APIDeployment.DoesNotExist:
+            raise Forbidden("API not found.")
+
+    @staticmethod
+    def list_api_keys_of_api(api_instance: APIDeployment) -> list[APIKey]:
+        api_keys: list[APIKey] = APIKey.objects.filter(api=api_instance).all()
+        return api_keys
+
+    @staticmethod
+    def has_access(api_key: APIKey, api_instance: APIDeployment) -> bool:
+        """Check if the provided API key has access to the specified API
+        instance.
+
+        Args:
+            api_key (APIKey): api key associated with the api
+            api_instance (APIDeployment): api model
+
+        Returns:
+            bool: True if allowed to execute, False otherwise
+        """
+        if not api_key.is_active:
+            return False
+        if isinstance(api_key.api, APIDeployment):
+            return api_key.api == api_instance
+        return False
+
+    @staticmethod
+    def validate_workflow_exists(workflow_id: str) -> None:
+        """Validate that the specified workflow_id exists in the Workflow
+        model."""
+        WorkflowHelper.get_workflow_by_id(workflow_id)
+
+    @staticmethod
+    def create_api_key(deployment: APIDeployment) -> APIKey:
+        """Create an APIKey entity with the data from the provided
+        APIDeployment instance."""
+        # Create an instance of the APIKey model
+        api_key_serializer = APIKeySerializer(
+            data={"api": deployment.id, "description": "Initial Access Key"},
+            context={"deployment": deployment},
+        )
+        api_key_serializer.is_valid(raise_exception=True)
+        api_key: APIKey = api_key_serializer.save()
+        return api_key
diff --git a/backend/api_v2/models.py b/backend/api_v2/models.py
new file mode 100644
index 0000000000..5454265370
--- /dev/null
+++ b/backend/api_v2/models.py
@@ -0,0 +1,167 @@
+import uuid
+from typing import Any
+
+from account_v2.models import User
+from api_v2.constants import ApiExecution
+from django.db import models
+from utils.models.base_model import BaseModel
+from utils.models.organization_mixin import (
+    DefaultOrganizationManagerMixin,
+    DefaultOrganizationMixin,
+)
+from utils.user_context import UserContext
+from workflow_manager.workflow_v2.models.workflow import Workflow
+
+API_NAME_MAX_LENGTH = 30
+DESCRIPTION_MAX_LENGTH = 255
+API_ENDPOINT_MAX_LENGTH = 255
+
+
+class APIDeploymentModelManager(DefaultOrganizationManagerMixin, models.Manager):
+    pass
+
+
+class APIDeployment(DefaultOrganizationMixin, BaseModel):
+    id = models.UUIDField(primary_key=True, default=uuid.uuid4, editable=False)
+    display_name = models.CharField(
+        max_length=API_NAME_MAX_LENGTH,
+        default="default api",
+        db_comment="User-given display name for the API.",
+    )
+    description = models.CharField(
+        max_length=DESCRIPTION_MAX_LENGTH,
+        blank=True,
+        default="",
+        db_comment="User-given description for the API.",
+    )
+    workflow = models.ForeignKey(
+        Workflow,
+        on_delete=models.CASCADE,
+        db_comment="Foreign key reference to the Workflow model.",
+        related_name="apis",
+    )
+    is_active = models.BooleanField(
+        default=True,
+        db_comment="Flag indicating whether the API is active or not.",
+    )
+    api_endpoint = models.CharField(
+        max_length=API_ENDPOINT_MAX_LENGTH,
+        unique=True,
+        editable=False,
+        db_comment="URL endpoint for the API deployment.",
+    )
+    api_name = models.CharField(
+        max_length=API_NAME_MAX_LENGTH,
+        default=uuid.uuid4,
+        db_comment="Short name for the API deployment.",
+    )
+    created_by = models.ForeignKey(
+        User,
+        on_delete=models.SET_NULL,
+        related_name="apis_created",
+        null=True,
+        blank=True,
+        editable=False,
+    )
+    modified_by = models.ForeignKey(
+        User,
+        on_delete=models.SET_NULL,
+        related_name="apis_modified",
+        null=True,
+        blank=True,
+        editable=False,
+    )
+
+    # Manager
+    objects = APIDeploymentModelManager()
+
+    def __str__(self) -> str:
+        return f"{self.id} - {self.display_name}"
+
+    def save(self, *args: Any, **kwargs: Any) -> None:
+        """Save hook to update api_endpoint.
+
+        Custom save hook for updating the 'api_endpoint' based on
+        'api_name'.     If the instance is being updated, it checks for
+        changes in 'api_name'         and adjusts 'api_endpoint'
+        accordingly.     If the instance is new, 'api_endpoint' is set
+        based on 'api_name'         and the current database schema.
+        """
+        if self.pk is not None:
+            organization_id = UserContext.get_organization_identifier()
+            try:
+                original = APIDeployment.objects.get(pk=self.pk)
+                if original.api_name != self.api_name:
+                    self.api_endpoint = (
+                        f"{ApiExecution.PATH}/{organization_id}/{self.api_name}/"
+                    )
+            except APIDeployment.DoesNotExist:
+                self.api_endpoint = (
+                    f"{ApiExecution.PATH}/{organization_id}/{self.api_name}/"
+                )
+        super().save(*args, **kwargs)
+
+    class Meta:
+        verbose_name = "Api Deployment"
+        verbose_name_plural = "Api Deployments"
+        db_table = "api_deployment_v2"
+        constraints = [
+            models.UniqueConstraint(
+                fields=["api_name", "organization"],
+                name="unique_api_name",
+            ),
+        ]
+
+
+class APIKey(BaseModel):
+    id = models.UUIDField(
+        primary_key=True,
+        editable=False,
+        default=uuid.uuid4,
+        db_comment="Unique identifier for the API key.",
+    )
+    api_key = models.UUIDField(
+        default=uuid.uuid4,
+        editable=False,
+        unique=True,
+        db_comment="Actual key UUID.",
+    )
+    api = models.ForeignKey(
+        APIDeployment,
+        on_delete=models.CASCADE,
+        db_comment="Foreign key reference to the APIDeployment model.",
+        related_name="api_keys",
+    )
+    description = models.CharField(
+        max_length=DESCRIPTION_MAX_LENGTH,
+        null=True,
+        db_comment="Description of the API key.",
+    )
+    is_active = models.BooleanField(
+        default=True,
+        db_comment="Flag indicating whether the API key is active or not.",
+    )
+    created_by = models.ForeignKey(
+        User,
+        on_delete=models.SET_NULL,
+        related_name="api_keys_created",
+        null=True,
+        blank=True,
+        editable=False,
+    )
+    modified_by = models.ForeignKey(
+        User,
+        on_delete=models.SET_NULL,
+        related_name="api_keys_modified",
+        null=True,
+        blank=True,
+        editable=False,
+    )
+
+    def __str__(self) -> str:
+        return f"{self.api.api_name} - {self.id} - {self.api_key}"
+
+    class Meta:
+        verbose_name = "Api Deployment key"
+        verbose_name_plural = "Api Deployment keys"
+        db_table = "api_deployment_key_v2"
diff --git a/backend/api_v2/postman_collection/__init__.py b/backend/api_v2/postman_collection/__init__.py
new file mode 100644
index 0000000000..e69de29bb2
diff --git a/backend/api_v2/postman_collection/constants.py b/backend/api_v2/postman_collection/constants.py
new file mode 100644
index 0000000000..ffb3ef729b
--- /dev/null
+++ b/backend/api_v2/postman_collection/constants.py
@@ -0,0 +1,6 @@
+class CollectionKey:
+    POSTMAN_COLLECTION_V210 = "https://schema.getpostman.com/json/collection/v2.1.0/collection.json"  # noqa: E501
+    EXECUTE_API_KEY = "Process document"
+    STATUS_API_KEY = "Execution status"
+    STATUS_EXEC_ID_DEFAULT = "REPLACE_WITH_EXECUTION_ID"
+    AUTH_QUERY_PARAM_DEFAULT = "REPLACE_WITH_API_KEY"
diff --git a/backend/api_v2/postman_collection/dto.py b/backend/api_v2/postman_collection/dto.py
new file mode 100644
index 0000000000..08f2294f19
--- /dev/null
+++ b/backend/api_v2/postman_collection/dto.py
@@ -0,0 +1,139 @@
+from dataclasses import asdict, dataclass, field
+from typing import Any, Optional
+from urllib.parse import urlencode, urljoin
+
+from api_v2.constants import ApiExecution
+from api_v2.models import APIDeployment
+from api_v2.postman_collection.constants import CollectionKey
+from django.conf import settings
+from utils.request import HTTPMethod
+
+
+@dataclass
+class HeaderItem:
+    key: str
+    value: str
+
+
+@dataclass
+class FormDataItem:
+    key: str
+    type: str
+    src: Optional[str] = None
+    value: Optional[str] = None
+
+    def __post_init__(self) -> None:
+        if self.type == "file":
+            if self.src is None:
+                raise ValueError("src must be provided for type 'file'")
+        elif self.type == "text":
+            if self.value is None:
+                raise ValueError("value must be provided for type 'text'")
+        else:
+            raise ValueError(f"Unsupported type for form data: {self.type}")
+
+
+@dataclass
+class BodyItem:
+    formdata: list[FormDataItem]
+    mode: str = "formdata"
+
+
+@dataclass
+class RequestItem:
+    method: HTTPMethod
+    url: str
+    header: list[HeaderItem]
+    body: Optional[BodyItem] = None
+
+
+@dataclass
+class PostmanItem:
+    name: str
+    request: RequestItem
+
+
+@dataclass
+class PostmanInfo:
+    name: str = "Unstract's API deployment"
+    schema: str = CollectionKey.POSTMAN_COLLECTION_V210
+    description: str = "Contains APIs meant for using the deployed Unstract API"
+
+
+@dataclass
+class PostmanCollection:
+    info: PostmanInfo
+    item: list[PostmanItem] = field(default_factory=list)
+
+    @classmethod
+    def create(
+        cls,
+        instance: APIDeployment,
+        api_key: str = CollectionKey.AUTH_QUERY_PARAM_DEFAULT,
+    ) -> "PostmanCollection":
+        """Creates a PostmanCollection instance.
+
+        This instance can help represent Postman collections (v2 format) that
+        can be used to easily invoke workflows deployed as APIs
+
+        Args:
+            instance (APIDeployment): API deployment to generate collection for
+            api_key (str, optional): Active API key used to authenticate requests for
+                deployed APIs. Defaults to CollectionKey.AUTH_QUERY_PARAM_DEFAULT.
+
+        Returns:
+            PostmanCollection: Instance representing PostmanCollection
+        """
+        postman_info = PostmanInfo(
+            name=instance.display_name, description=instance.description
+        )
+        header_list = [HeaderItem(key="Authorization", value=f"Bearer {api_key}")]
+        abs_api_endpoint = urljoin(settings.WEB_APP_ORIGIN_URL, instance.api_endpoint)
+
+        # API execution API
+        execute_body = BodyItem(
+            formdata=[
+                FormDataItem(
+                    key=ApiExecution.FILES_FORM_DATA, type="file", src="/path_to_file"
+                ),
+                FormDataItem(
+                    key=ApiExecution.TIMEOUT_FORM_DATA,
+                    type="text",
+                    value=ApiExecution.MAXIMUM_TIMEOUT_IN_SEC,
+                ),
+                FormDataItem(
+                    key=ApiExecution.INCLUDE_METADATA,
+                    type="text",
+                    value=False,
+                ),
+            ]
+        )
+        execute_request = RequestItem(
+            method=HTTPMethod.POST,
+            header=header_list,
+            body=execute_body,
+            url=abs_api_endpoint,
+        )
+
+        # Status API
+        status_query_param = {"execution_id": CollectionKey.STATUS_EXEC_ID_DEFAULT}
+        status_query_str = urlencode(status_query_param)
+        status_url = urljoin(abs_api_endpoint, "?" + status_query_str)
+        status_request = RequestItem(
+            method=HTTPMethod.GET, header=header_list, url=status_url
+        )
+
+        postman_item_list = [
+            PostmanItem(name=CollectionKey.EXECUTE_API_KEY, request=execute_request),
+            PostmanItem(name=CollectionKey.STATUS_API_KEY, request=status_request),
+        ]
+        return cls(info=postman_info, item=postman_item_list)
+
+    def to_dict(self) -> dict[str, Any]:
+        """Convert PostmanCollection instance to a dict.
+
+        Returns:
+            dict[str, Any]: PostmanCollection as a dict
+        """
+        collection_dict = asdict(self)
+        return collection_dict
diff --git a/backend/api_v2/serializers.py b/backend/api_v2/serializers.py
new file mode 100644
index 0000000000..522e5a23ed
--- /dev/null
+++ b/backend/api_v2/serializers.py
@@ -0,0 +1,123 @@
+from collections import OrderedDict
+from typing import Any, Union
+
+from api_v2.constants import ApiExecution
+from api_v2.models import APIDeployment, APIKey
+from django.core.validators import RegexValidator
+from rest_framework.serializers import (
+    CharField,
+    IntegerField,
+    JSONField,
+    ModelSerializer,
+    Serializer,
+    ValidationError,
+)
+
+from backend.serializers import AuditSerializer
+
+
+class APIDeploymentSerializer(AuditSerializer):
+    class Meta:
+        model = APIDeployment
+        fields = "__all__"
+
+    def validate_api_name(self, value: str) -> str:
+        api_name_validator = RegexValidator(
+            regex=r"^[a-zA-Z0-9_-]+$",
+            message="Only letters, numbers, hyphen and \
+                underscores are allowed.",
+            code="invalid_api_name",
+        )
+        api_name_validator(value)
+        return value
+
+
+class APIKeySerializer(AuditSerializer):
+    class Meta:
+        model = APIKey
+        fields = "__all__"
+
+    def to_representation(self, instance: APIKey) -> OrderedDict[str, Any]:
+        """Override the to_representation method to include additional
+        context."""
+        context = self.context.get("context", {})
+        deployment: APIDeployment = context.get("deployment")
+        representation: OrderedDict[str, Any] = super().to_representation(instance)
+        if deployment:
+            representation["api"] = deployment.id
+            representation["description"] = f"API Key for {deployment.name}"
+            representation["is_active"] = True
+
+        return representation
+
+
+class ExecutionRequestSerializer(Serializer):
+    """Execution request serializer
+    timeout: 0: maximum value of timeout, -1: async execution
+    """
+
+    timeout = IntegerField(
+        min_value=-1, max_value=ApiExecution.MAXIMUM_TIMEOUT_IN_SEC, default=-1
+    )
+
+    def validate_timeout(self, value: Any) -> int:
+        if not isinstance(value, int):
+            raise ValidationError("timeout must be a integer.")
+        if value == 0:
+            value = ApiExecution.MAXIMUM_TIMEOUT_IN_SEC
+        return value
+
+    def get_timeout(self, validated_data: dict[str, Union[int, None]]) -> int:
+        value = validated_data.get(ApiExecution.TIMEOUT_FORM_DATA, -1)
+        if not isinstance(value, int):
+            raise ValidationError("timeout must be a integer.")
+        return value
+
+
+class APIDeploymentListSerializer(ModelSerializer):
+    workflow_name = CharField(source="workflow.workflow_name", read_only=True)
+
+    class Meta:
+        model = APIDeployment
+        fields = [
+            "id",
+            "workflow",
+            "workflow_name",
+            "display_name",
+            "description",
+            "is_active",
+            "api_endpoint",
+            "api_name",
+            "created_by",
+        ]
+
+
+class APIKeyListSerializer(ModelSerializer):
+    class Meta:
+        model = APIKey
+        fields = [
+            "id",
+            "created_at",
+            "modified_at",
+            "api_key",
+            "is_active",
+            "description",
+            "api",
+        ]
+
+
+class DeploymentResponseSerializer(Serializer):
+    is_active = CharField()
+    id = CharField()
+    api_key = CharField()
+    api_endpoint = CharField()
+    display_name = CharField()
+    description = CharField()
+    api_name = CharField()
+
+
+class APIExecutionResponseSerializer(Serializer):
+    execution_status = CharField()
+    status_api = CharField()
+    error = CharField()
+    result = JSONField()
diff --git a/backend/api_v2/urls.py b/backend/api_v2/urls.py
new file mode 100644
index 0000000000..065617d666
--- /dev/null
+++ b/backend/api_v2/urls.py
@@ -0,0 +1,63 @@
+from api_v2.api_deployment_views import APIDeploymentViewSet, DeploymentExecution
+from api_v2.api_key_views import APIKeyViewSet
+from django.urls import path, re_path
+from rest_framework.urlpatterns import format_suffix_patterns
+
+deployment = APIDeploymentViewSet.as_view(
+    {
+        "get": APIDeploymentViewSet.list.__name__,
+        "post": APIDeploymentViewSet.create.__name__,
+    }
+)
+deployment_details = APIDeploymentViewSet.as_view(
+    {
+        "get": APIDeploymentViewSet.retrieve.__name__,
+        "put": APIDeploymentViewSet.update.__name__,
+        "patch": APIDeploymentViewSet.partial_update.__name__,
+        "delete": APIDeploymentViewSet.destroy.__name__,
+    }
+)
+download_postman_collection = APIDeploymentViewSet.as_view(
+    {
+        "get": APIDeploymentViewSet.download_postman_collection.__name__,
+    }
+)
+
+execute = DeploymentExecution.as_view()
+
+key_details = APIKeyViewSet.as_view(
+    {
+        "get": APIKeyViewSet.retrieve.__name__,
+        "put": APIKeyViewSet.update.__name__,
+        "delete": APIKeyViewSet.destroy.__name__,
+    }
+)
+api_key = APIKeyViewSet.as_view(
+    {
+        "get": APIKeyViewSet.api_keys.__name__,
+        "post": APIKeyViewSet.create.__name__,
+    }
+)
+
+urlpatterns = format_suffix_patterns(
+    [
+        path("deployment/", deployment, name="api_deployment"),
+        path(
+            "deployment/<uuid:pk>/",
+            deployment_details,
+            name="api_deployment_details",
+        ),
+        path(
+            "postman_collection/<uuid:pk>/",
+            download_postman_collection,
+            name="download_postman_collection",
+        ),
+        re_path(
+            r"^api/(?P<org_name>[\w-]+)/(?P<api_name>[\w-]+)/?$",
+            execute,
+            name="api_deployment_execution",
+        ),
+        path("keys/<uuid:pk>/", key_details, name="key_details"),
+        path("keys/api/<str:api_id>/", api_key, name="api_key"),
+    ]
+)
diff --git a/backend/pipeline_v2/__init__.py b/backend/pipeline_v2/__init__.py
new file mode 100644
index 0000000000..e69de29bb2
diff --git a/backend/pipeline_v2/constants.py b/backend/pipeline_v2/constants.py
new file mode 100644
index 0000000000..2475ad79f0
--- /dev/null
+++ b/backend/pipeline_v2/constants.py
@@ -0,0 +1,66 @@
+class PipelineConstants:
+    """Constants for Pipelines."""
+
+    TYPE = "type"
+    ETL_PIPELINE = "ETL"
+    TASK_PIPELINE = "TASK"
+    ETL = "etl"
+    TASK = "task"
+    CREATE_ACTION = "create"
+    UPDATE_ACTION = "update"
+    PIPELINE_GUID = "id"
+    ACTION = "action"
+    NOT_CONFIGURED = "Connector not configured."
+    SOURCE_NOT_CONFIGURED = "Source not configured."
+    DESTINATION_NOT_CONFIGURED = "Destination not configured."
+    SOURCE_ICON = "source_icon"
+    DESTINATION_ICON = "destination_icon"
+    SOURCE_NAME = "source_name"
+    DESTINATION_NAME = "destination_name"
+    INPUT_FILE = "input_file_connector"
+    INPUT_DB = "input_db_connector"
+    OUTPUT_FILE = "output_file_connector"
+    OUTPUT_DB = "output_db_connector"
+    SOURCE = "source"
+    DEST = "dest"
+
+
+class PipelineExecutionKey:
+    PIPELINE = "pipeline"
+    EXECUTION = "execution"
+
+
+class PipelineKey:
+    """Constants for the Pipeline model."""
+
+    PIPELINE_GUID = "id"
+    PIPELINE_NAME = "pipeline_name"
+    WORKFLOW = "workflow"
+    APP_ID = "app_id"
+    ACTIVE = "active"
+    SCHEDULED = "scheduled"
+    PIPELINE_TYPE = "pipeline_type"
+    RUN_COUNT = "run_count"
+    LAST_RUN_TIME = "last_run_time"
+    LAST_RUN_STATUS = "last_run_status"
+    # Used by serializer
+    CRON_DATA = "cron_data"
+    WORKFLOW_NAME = "workflow_name"
+    WORKFLOW_ID = "workflow_id"
+    CRON_STRING = "cron_string"
+    PIPELINE_ID = "pipeline_id"
+
+
+class PipelineErrors:
+    PIPELINE_EXISTS = "Pipeline with this configuration might already exist or some mandatory field is missing."  # noqa: E501
+    DUPLICATE_API = "It appears that a duplicate call may have been made."
+    INVALID_WF = "The provided workflow does not exist"
+
+
+class PipelineURL:
+    """Constants for URL names."""
+
+    DETAIL = "pipeline-detail"
+    EXECUTIONS = "pipeline-executions"
+    LIST = "pipeline-list"
+    EXECUTE = "tenant:pipeline-execute"
diff --git a/backend/pipeline_v2/exceptions.py b/backend/pipeline_v2/exceptions.py
new file mode 100644
index 0000000000..8da26c481f
--- /dev/null
+++ b/backend/pipeline_v2/exceptions.py
@@ -0,0 +1,46 @@
+from typing import Optional
+
+from rest_framework.exceptions import APIException
+
+
+class WorkflowTriggerError(APIException):
+    status_code = 400
+    default_detail = "Error triggering workflow. Pipeline created"
+
+
+class PipelineExecuteError(APIException):
+    status_code = 500
+    default_detail = "Error executing pipline"
+
+
+class InactivePipelineError(APIException):
+    status_code = 422
+    default_detail = "Pipeline is inactive, please activate the pipeline"
+
+    def __init__(
+        self,
+        pipeline_name: Optional[str] = None,
+        detail: Optional[str] = None,
+        code: Optional[str] = None,
+    ):
+        if pipeline_name:
+            self.default_detail = (
+                f"Pipeline '{pipeline_name}' is inactive, "
+                "please activate the pipeline"
+            )
+        super().__init__(detail, code)
+
+
+class MandatoryPipelineType(APIException):
+    status_code = 400
+    default_detail = "Pipeline type is mandatory"
+
+
+class MandatoryWorkflowId(APIException):
+    status_code = 400
+    default_detail = "Workflow ID is mandatory"
+
+
+class MandatoryCronSchedule(APIException):
+    status_code = 400
+    default_detail = "Cron schedule is mandatory"
diff --git a/backend/pipeline_v2/execution_view.py b/backend/pipeline_v2/execution_view.py
new file mode 100644
index 0000000000..fa514615aa
--- /dev/null
+++ b/backend/pipeline_v2/execution_view.py
@@ -0,0 +1,35 @@
+from permissions.permission import IsOwner
+from pipeline_v2.serializers.execute import DateRangeSerializer
+from rest_framework import viewsets
+from rest_framework.versioning import URLPathVersioning
+from utils.pagination import CustomPagination
+from workflow_manager.workflow_v2.models.execution import WorkflowExecution
+from workflow_manager.workflow_v2.serializers import WorkflowExecutionSerializer
+
+
+class PipelineExecutionViewSet(viewsets.ModelViewSet):
+    versioning_class = URLPathVersioning
+    permission_classes = [IsOwner]
+    serializer_class = WorkflowExecutionSerializer
+    pagination_class = CustomPagination
+
+    CREATED_AT_FIELD_DESC = "-created_at"
+    START_DATE_FIELD = "start_date"
+    END_DATE_FIELD = "end_date"
+
+    def get_queryset(self):
+        # Get the pipeline_id from the URL path
+        pipeline_id = self.kwargs.get("pk")
+        queryset = WorkflowExecution.objects.filter(pipeline_id=pipeline_id)
+
+        # Validate start_date and end_date parameters using DateRangeSerializer
+        date_range_serializer = DateRangeSerializer(data=self.request.query_params)
+        date_range_serializer.is_valid(raise_exception=True)
+        start_date = date_range_serializer.validated_data.get(self.START_DATE_FIELD)
+        end_date = date_range_serializer.validated_data.get(self.END_DATE_FIELD)
+
+        if start_date and end_date:
+            queryset = queryset.filter(created_at__range=(start_date, end_date))
+
+        queryset = queryset.order_by(self.CREATED_AT_FIELD_DESC)
+        return queryset
diff --git a/backend/pipeline_v2/manager.py b/backend/pipeline_v2/manager.py
new file mode 100644
index 0000000000..d5297107d7
--- /dev/null
+++ b/backend/pipeline_v2/manager.py
@@ -0,0 +1,59 @@
+import logging
+from typing import Any, Optional
+
+from django.conf import settings
+from django.urls import reverse
+from pipeline_v2.constants import PipelineKey, PipelineURL
+from pipeline_v2.models import Pipeline
+from pipeline_v2.pipeline_processor import PipelineProcessor
+from rest_framework.request import Request
+from rest_framework.response import Response
+from utils.request.constants import RequestConstants
+from workflow_manager.workflow_v2.constants import WorkflowExecutionKey, WorkflowKey
+from workflow_manager.workflow_v2.views import WorkflowViewSet
+
+from backend.constants import RequestHeader
+
+logger = logging.getLogger(__name__)
+
+
+class PipelineManager:
+    """Helps manage the execution and scheduling of pipelines."""
+
+    @staticmethod
+    def execute_pipeline(
+        request: Request,
+        pipeline_id: str,
+        execution_id: Optional[str] = None,
+    ) -> Response:
+        """Used to execute a pipeline.
+
+        Args:
+            pipeline_id (str): UUID of the pipeline to execute
+            execution_id (Optional[str], optional):
+                Uniquely identifies an execution. Defaults to None.
+        """
+        logger.info(f"Executing pipeline {pipeline_id}, execution: {execution_id}")
+        pipeline: Pipeline = PipelineProcessor.initialize_pipeline_sync(pipeline_id)
+        # TODO: Use DRF's request and as_view() instead
+        request.data[WorkflowKey.WF_ID] = pipeline.workflow.id
+        if execution_id is not None:
+            request.data[WorkflowExecutionKey.EXECUTION_ID] = execution_id
+        wf_viewset = WorkflowViewSet()
+        return wf_viewset.execute(request=request, pipeline_guid=str(pipeline.pk))
+
+    @staticmethod
+    def get_pipeline_execution_data_for_scheduled_run(
+        pipeline_id: str,
+    ) -> Optional[dict[str, Any]]:
+        """Gets the required data to be passed while executing a pipeline Any
+        changes to pipeline execution needs to be propagated here."""
+        callback_url = settings.DJANGO_APP_BACKEND_URL + reverse(PipelineURL.EXECUTE)
+        job_headers = {RequestHeader.X_API_KEY: settings.INTERNAL_SERVICE_API_KEY}
+        job_kwargs = {
+            RequestConstants.VERB: "POST",
+            RequestConstants.URL: callback_url,
+            RequestConstants.HEADERS: job_headers,
+            RequestConstants.DATA: {PipelineKey.PIPELINE_ID: pipeline_id},
+        }
+        return job_kwargs
diff --git a/backend/pipeline_v2/models.py b/backend/pipeline_v2/models.py
new file mode 100644
index 0000000000..35ef1914c0
--- /dev/null
+++ b/backend/pipeline_v2/models.py
@@ -0,0 +1,111 @@
+import uuid
+
+from account_v2.models import User
+from django.db import models
+from utils.models.base_model import BaseModel
+from utils.models.organization_mixin import (
+    DefaultOrganizationManagerMixin,
+    DefaultOrganizationMixin,
+)
+from workflow_manager.workflow_v2.models.workflow import Workflow
+
+from backend.constants import FieldLengthConstants as FieldLength
+
+APP_ID_LENGTH = 32
+PIPELINE_NAME_LENGTH = 32
+
+
+class PipelineModelManager(DefaultOrganizationManagerMixin, models.Manager):
+    pass
+
+
+class Pipeline(DefaultOrganizationMixin, BaseModel):
+    """Model to hold data related to Pipelines."""
+
+    class PipelineType(models.TextChoices):
+        ETL = "ETL", "ETL"
+        TASK = "TASK", "TASK"
+        DEFAULT = "DEFAULT", "Default"
+        APP = "APP", "App"
+
+    class PipelineStatus(models.TextChoices):
+        SUCCESS = "SUCCESS", "Success"
+        FAILURE = "FAILURE", "Failure"
+        INPROGRESS = "INPROGRESS", "Inprogress"
+        YET_TO_START = "YET_TO_START", "Yet to start"
+        RESTARTING = "RESTARTING", "Restarting"
+        PAUSED = "PAUSED", "Paused"
+
+    id = models.UUIDField(primary_key=True, default=uuid.uuid4, editable=False)
+    pipeline_name = models.CharField(max_length=PIPELINE_NAME_LENGTH, default="")
+    workflow = models.ForeignKey(
+        Workflow,
+        on_delete=models.CASCADE,
+        related_name="pipelines",
+        null=False,
+        blank=False,
+    )
+    # Added as text field until a model for App is included.
+    app_id = models.TextField(null=True, blank=True, max_length=APP_ID_LENGTH)
+    active = models.BooleanField(default=False)  # TODO: Add dbcomment
+    scheduled = models.BooleanField(default=False)  # TODO: Add dbcomment
+    cron_string = models.TextField(
+        db_comment="UNIX cron string",
+        null=False,
+        blank=False,
+        max_length=FieldLength.CRON_LENGTH,
+    )
+    pipeline_type = models.CharField(
+        choices=PipelineType.choices, default=PipelineType.DEFAULT
+    )
+    run_count = models.IntegerField(default=0)
+    last_run_time = models.DateTimeField(null=True, blank=True)
+    last_run_status = models.CharField(
+        choices=PipelineStatus.choices, default=PipelineStatus.YET_TO_START
+    )
+    app_icon = models.URLField(
+        null=True, blank=True, db_comment="Field to store icon url for Apps"
+    )
+    app_url = models.URLField(
+        null=True, blank=True, db_comment="Stores deployed URL for App"
+    )
+    # TODO: Change this to a Forgein key once the bundle is created.
+    access_control_bundle_id = models.TextField(null=True, blank=True)
+    created_by = models.ForeignKey(
+        User,
+        on_delete=models.SET_NULL,
+        related_name="pipelines_created",
+        null=True,
+        blank=True,
+    )
+    modified_by = models.ForeignKey(
+        User,
+        on_delete=models.SET_NULL,
+        related_name="pipelines_modified",
+        null=True,
+        blank=True,
+    )
+
+    # Manager
+    objects = PipelineModelManager()
+
+    def __str__(self) -> str:
+        return f"Pipeline({self.id})"
+
+    class Meta:
+        verbose_name = "Pipeline"
+        verbose_name_plural = "Pipelines"
+        db_table = "pipeline_v2"
+        constraints = [
+            models.UniqueConstraint(
+                fields=["id", "pipeline_type"],
+                name="unique_pipeline_entity",
+            ),
+            models.UniqueConstraint(
+                fields=["pipeline_name", "organization"],
+                name="unique_pipeline_name",
+            ),
+        ]
+
+    def is_active(self) -> bool:
+        return bool(self.active)
diff --git a/backend/pipeline_v2/pipeline_processor.py b/backend/pipeline_v2/pipeline_processor.py
new file mode 100644
index 0000000000..fa3328db2f
--- /dev/null
+++ b/backend/pipeline_v2/pipeline_processor.py
@@ -0,0 +1,81 @@
+import logging
+from typing import Optional
+
+from django.utils import timezone
+from pipeline_v2.exceptions import InactivePipelineError
+from pipeline_v2.models import Pipeline
+
+logger = logging.getLogger(__name__)
+
+
+class PipelineProcessor:
+    @staticmethod
+    def initialize_pipeline_sync(pipeline_id: str) -> Pipeline:
+        """Fetches and initializes the sync for a pipeline.
+
+        Args:
+            pipeline_id (str): UUID of the pipeline to sync
+        """
+        pipeline: Pipeline = PipelineProcessor.fetch_pipeline(pipeline_id)
+        pipeline.run_count = pipeline.run_count + 1
+        return PipelineProcessor._update_pipeline_status(
+            pipeline=pipeline,
+            status=Pipeline.PipelineStatus.RESTARTING,
+            is_end=False,
+        )
+
+    @staticmethod
+    def fetch_pipeline(pipeline_id: str, check_active: bool = True) -> Pipeline:
+        """Retrieves and checks for an active pipeline.
+
+        Raises:
+            InactivePipelineError: If an active pipeline is not found
+        """
+        pipeline: Pipeline = Pipeline.objects.get(pk=pipeline_id)
+        if check_active and not pipeline.is_active():
+            logger.error(f"Inactive pipeline fetched: {pipeline_id}")
+            raise InactivePipelineError(pipeline_name=pipeline.pipeline_name)
+        return pipeline
+
+    @staticmethod
+    def _update_pipeline_status(
+        pipeline: Pipeline,
+        status: tuple[str, str],
+        is_end: bool,
+        is_active: Optional[bool] = None,
+    ) -> Pipeline:
+        """Updates pipeline status during execution.
+
+        Raises:
+            PipelineSaveError: Exception while saving a pipeline
+
+        Returns:
+            Pipeline: Updated pipeline
+        """
+        if is_end:
+            pipeline.last_run_time = timezone.now()
+        if status:
+            pipeline.last_run_status = status
+        if is_active is not None:
+            pipeline.active = is_active
+
+        pipeline.save()
+        return pipeline
+
+    @staticmethod
+    def update_pipeline(
+        pipeline_guid: Optional[str],
+        status: tuple[str, str],
+        is_active: Optional[bool] = None,
+    ) -> None:
+        if not pipeline_guid:
+            return
+        # Skip check if we are enabling an inactive pipeline
+        check_active = not is_active
+        pipeline: Pipeline = PipelineProcessor.fetch_pipeline(
+            pipeline_id=pipeline_guid, check_active=check_active
+        )
+        PipelineProcessor._update_pipeline_status(
+            pipeline=pipeline, is_end=True, status=status, is_active=is_active
+        )
+        logger.info(f"Updated pipeline {pipeline_guid} status: {status}")
diff --git a/backend/pipeline_v2/serializers/crud.py b/backend/pipeline_v2/serializers/crud.py
new file mode 100644
index 0000000000..bb051abef0
--- /dev/null
+++ b/backend/pipeline_v2/serializers/crud.py
@@ -0,0 +1,104 @@
+import logging
+from collections import OrderedDict
+from typing import Any
+
+from connector_processor.connector_processor import ConnectorProcessor
+from connector_v2.connector_instance_helper import ConnectorInstanceHelper
+from connector_v2.models import ConnectorInstance
+from pipeline_v2.constants import PipelineConstants as PC
+from pipeline_v2.constants import PipelineKey as PK
+from pipeline_v2.models import Pipeline
+from scheduler.helper import SchedulerHelper
+from utils.serializer_utils import SerializerUtils
+
+from backend.serializers import AuditSerializer
+from unstract.connectors.connectorkit import Connectorkit
+
+logger = logging.getLogger(__name__)
+
+
+class PipelineSerializer(AuditSerializer):
+
+    class Meta:
+        model = Pipeline
+        fields = "__all__"
+
+    def create(self, validated_data: dict[str, Any]) -> Any:
+        # TODO: Deduce pipeline type based on WF?
+        validated_data[PK.ACTIVE] = True  # Add this as default instead?
+        validated_data[PK.SCHEDULED] = True
+        return super().create(validated_data)
+
+    def save(self, **kwargs: Any) -> Pipeline:
+        pipeline: Pipeline = super().save(**kwargs)
+        SchedulerHelper.add_job(
+            str(pipeline.pk),
+            cron_string=pipeline.cron_string,
+        )
+        return pipeline
+
+    def _get_name_and_icon(self, connectors: list[Any], connector_id: Any) -> Any:
+        for obj in connectors:
+            if obj["id"] == connector_id:
+                return obj["name"], obj["icon"]
+        return PC.NOT_CONFIGURED, None
+
+    def _add_connector_data(
+        self,
+        repr: OrderedDict[str, Any],
+        connector_instance_list: list[Any],
+        connectors: list[Any],
+    ) -> OrderedDict[str, Any]:
+        """Adds connector Input/Output data.
+
+        Args:
+            sef (_type_): _description_
+            repr (OrderedDict[str, Any]): _description_
+
+        Returns:
+            OrderedDict[str, Any]: _description_
+        """
+        repr[PC.SOURCE_NAME] = PC.NOT_CONFIGURED
+        repr[PC.DESTINATION_NAME] = PC.NOT_CONFIGURED
+        for instance in connector_instance_list:
+            if instance.connector_type == "INPUT":
+                repr[PC.SOURCE_NAME], repr[PC.SOURCE_ICON] = self._get_name_and_icon(
+                    connectors=connectors,
+                    connector_id=instance.connector_id,
+                )
+            if instance.connector_type == "OUTPUT":
+                repr[PC.DESTINATION_NAME], repr[PC.DESTINATION_ICON] = (
+                    self._get_name_and_icon(
+                        connectors=connectors,
+                        connector_id=instance.connector_id,
+                    )
+                )
+        return repr
+
+    def to_representation(self, instance: Pipeline) -> OrderedDict[str, Any]:
+        """To set Source, Destination & Agency for Pipelines."""
+        repr: OrderedDict[str, Any] = super().to_representation(instance)
+
+        connector_kit = Connectorkit()
+        connectors = connector_kit.get_connectors_list()
+
+        if SerializerUtils.check_context_for_GET_or_POST(context=self.context):
+            workflow = instance.workflow
+            connector_instance_list = ConnectorInstanceHelper.get_input_output_connector_instances_by_workflow(  # noqa
+                workflow.id
+            )
+            repr[PK.WORKFLOW_ID] = workflow.id
+            repr[PK.WORKFLOW_NAME] = workflow.workflow_name
+            repr[PK.CRON_STRING] = repr.pop(PK.CRON_STRING)
+            repr = self._add_connector_data(
+                repr=repr,
+                connector_instance_list=connector_instance_list,
+                connectors=connectors,
+            )
+
+        return repr
+
+    def get_connector_data(self, connector: ConnectorInstance, key: str) -> Any:
+        return ConnectorProcessor.get_connector_data_with_key(
+            connector.connector_id, key
+        )
diff --git a/backend/pipeline_v2/serializers/execute.py b/backend/pipeline_v2/serializers/execute.py
new file mode 100644
index 0000000000..4e5a7fd582
--- /dev/null
+++ b/backend/pipeline_v2/serializers/execute.py
@@ -0,0 +1,24 @@
+import logging
+
+from pipeline_v2.models import Pipeline
+from rest_framework import serializers
+
+logger = logging.getLogger(__name__)
+
+
+class PipelineExecuteSerializer(serializers.Serializer):
+    # TODO: Add pipeline as a read_only related field
+    pipeline_id = serializers.UUIDField()
+    execution_id = serializers.UUIDField(required=False)
+
+    def validate_pipeline_id(self, value: str) -> str:
+        try:
+            Pipeline.objects.get(pk=value)
+        except Pipeline.DoesNotExist:
+            raise serializers.ValidationError("Invalid pipeline ID")
+        return value
+
+
+class DateRangeSerializer(serializers.Serializer):
+    start_date = serializers.DateTimeField(required=False)
+    end_date = serializers.DateTimeField(required=False)
diff --git a/backend/pipeline_v2/serializers/update.py b/backend/pipeline_v2/serializers/update.py
new file mode 100644
index 0000000000..6fc9f66dc5
--- /dev/null
+++ b/backend/pipeline_v2/serializers/update.py
@@ -0,0 +1,14 @@
+from pipeline_v2.models import Pipeline
+from rest_framework import serializers
+
+
+class PipelineUpdateSerializer(serializers.Serializer):
+    pipeline_id = serializers.UUIDField(required=True)
+    active = serializers.BooleanField(required=True)
+
+    def validate_pipeline_id(self, value: str) -> str:
+        try:
+            Pipeline.objects.get(pk=value)
+        except Pipeline.DoesNotExist:
+            raise serializers.ValidationError("Invalid pipeline ID")
+        return value
diff --git a/backend/pipeline_v2/urls.py b/backend/pipeline_v2/urls.py
new file mode 100644
index 0000000000..d6950633db
--- /dev/null
+++ b/backend/pipeline_v2/urls.py
@@ -0,0 +1,41 @@
+from django.urls import path
+from pipeline_v2.constants import PipelineURL
+from pipeline_v2.execution_view import PipelineExecutionViewSet
+from pipeline_v2.views import PipelineViewSet
+from rest_framework.urlpatterns import format_suffix_patterns
+
+pipeline_list = PipelineViewSet.as_view(
+    {
+        "get": "list",
+        "post": "create",
+    }
+)
+execution_list = PipelineExecutionViewSet.as_view(
+    {
+        "get": "list",
+    }
+)
+pipeline_detail = PipelineViewSet.as_view(
+    {
+        "get": "retrieve",
+        "put": "update",
+        "patch": "partial_update",
+        "delete": "destroy",
+    }
+)
+
+pipeline_execute = PipelineViewSet.as_view({"post": "execute"})
+
+
+urlpatterns = format_suffix_patterns(
+    [
+        path("pipeline/", pipeline_list, name=PipelineURL.LIST),
+        path("pipeline/<uuid:pk>/", pipeline_detail, name=PipelineURL.DETAIL),
+        path(
+            "pipeline/<uuid:pk>/executions/",
+            execution_list,
+            name=PipelineURL.EXECUTIONS,
+        ),
+        path("pipeline/execute/", pipeline_execute, name=PipelineURL.EXECUTE),
+    ]
+)
diff --git a/backend/pipeline_v2/views.py b/backend/pipeline_v2/views.py
new file mode 100644
index 0000000000..553b0179a8
--- /dev/null
+++ b/backend/pipeline_v2/views.py
@@ -0,0 +1,118 @@
+import logging
+from typing import Any, Optional
+
+from account_v2.custom_exceptions import DuplicateData
+from django.db import IntegrityError
+from django.db.models import QuerySet
+from permissions.permission import IsOwner
+from pipeline_v2.constants import (
+    PipelineConstants,
+    PipelineErrors,
+    PipelineExecutionKey,
+)
+from pipeline_v2.constants import PipelineKey as PK
+from pipeline_v2.manager import PipelineManager
+from pipeline_v2.models import Pipeline
+from pipeline_v2.pipeline_processor import PipelineProcessor
+from pipeline_v2.serializers.crud import PipelineSerializer
+from pipeline_v2.serializers.execute import (
+    PipelineExecuteSerializer as ExecuteSerializer,
+)
+from pipeline_v2.serializers.update import PipelineUpdateSerializer
+from rest_framework import serializers, status, viewsets
+from rest_framework.request import Request
+from rest_framework.response import Response
+from rest_framework.versioning import URLPathVersioning
+from scheduler.helper import SchedulerHelper
+
+logger = logging.getLogger(__name__)
+
+
+class PipelineViewSet(viewsets.ModelViewSet):
+    versioning_class = URLPathVersioning
+    queryset = Pipeline.objects.all()
+    permission_classes = [IsOwner]
+    serializer_class = PipelineSerializer
+
+    def get_queryset(self) -> Optional[QuerySet]:
+        type = self.request.query_params.get(PipelineConstants.TYPE)
+        if type is not None:
+            queryset = Pipeline.objects.filter(
+                created_by=self.request.user, pipeline_type=type
+            )
+            return queryset
+        elif type is None:
+            queryset = Pipeline.objects.filter(created_by=self.request.user)
+            return queryset
+
+    def get_serializer_class(self) -> serializers.Serializer:
+        if self.action == "execute":
+            return ExecuteSerializer
+        else:
+            return PipelineSerializer
+
+    # TODO: Refactor to perform an action with explicit arguments
+    # For eg, passing pipeline ID and with_log=False -> executes pipeline
+    # For FE however we call the same API twice
+    # (first call generates execution ID)
+    def execute(self, request: Request) -> Response:
+        serializer: ExecuteSerializer = self.get_serializer(data=request.data)
+        serializer.is_valid(raise_exception=True)
+        execution_id = serializer.validated_data.get("execution_id", None)
+        pipeline_id = serializer.validated_data[PK.PIPELINE_ID]
+
+        execution = PipelineManager.execute_pipeline(
+            request=request,
+            pipeline_id=pipeline_id,
+            execution_id=execution_id,
+        )
+        pipeline: Pipeline = PipelineProcessor.fetch_pipeline(pipeline_id)
+        serializer = PipelineSerializer(pipeline)
+        response_data = {
+            PipelineExecutionKey.PIPELINE: serializer.data,
+            PipelineExecutionKey.EXECUTION: execution.data,
+        }
+        return Response(data=response_data, status=status.HTTP_200_OK)
+
+    def create(self, request: Request) -> Response:
+        serializer = self.get_serializer(data=request.data)
+        serializer.is_valid(raise_exception=True)
+        try:
+            serializer.save()
+        except IntegrityError:
+            raise DuplicateData(
+                f"{PipelineErrors.PIPELINE_EXISTS}, " f"{PipelineErrors.DUPLICATE_API}"
+            )
+        return Response(data=serializer.data, status=status.HTTP_201_CREATED)
+
+    def perform_destroy(self, instance: Pipeline) -> None:
+        pipeline_to_remove = str(instance.pk)
+        super().perform_destroy(instance)
+        return SchedulerHelper.remove_job(pipeline_to_remove)
+
+    def partial_update(self, request: Request, pk: Any = None) -> Response:
+        serializer = PipelineUpdateSerializer(data=request.data)
+        if serializer.is_valid():
+            pipeline_id = serializer.validated_data.get("pipeline_id")
+            active = serializer.validated_data.get("active")
+            try:
+                if active:
+                    SchedulerHelper.resume_job(pipeline_id)
+                else:
+                    SchedulerHelper.pause_job(pipeline_id)
+            except Exception as e:
+                logger.error(f"Failed to update pipeline status: {e}")
+                return Response(
+                    {"error": "Failed to update pipeline status"},
+                    status=status.HTTP_500_INTERNAL_SERVER_ERROR,
+                )
+
+            return Response(
+                {
+                    "status": "success",
+                    "message": f"Pipeline {pipeline_id} status updated",
+                },
+                status=status.HTTP_200_OK,
+            )
+        else:
+            return Response(serializer.errors, status=status.HTTP_400_BAD_REQUEST)