async def create_source_connection(
        self,
        db: AsyncSession,
        source_connection_in: schemas.SourceConnectionCreate,
        current_user: schemas.User,
    ) -> Tuple[schemas.SourceConnection, Optional[schemas.SyncJob]]:
        """Create a new source connection with all related objects.

        This method:
        1. Creates a credential with auth fields if provided, or uses existing credential
        2. Creates the connection to the source (schemas.Connection)
        3. Creates a collection if not provided
        4. Creates a sync configuration and DAG
        5. Creates a sync job if immediate execution is requested
        6. Creates the source connection (schemas.SourceConnection)

        Args:
            db: The database session
            source_connection_in: The source connection to create
            current_user: The current user

        Returns:
            A tuple of (source_connection, sync_job)

        Raises:
            HTTPException: If the source is not found
        """
        # Separate core and auxiliary attributes
        core_attrs, aux_attrs = source_connection_in.map_to_core_and_auxiliary_attributes()

        async with UnitOfWork(db) as uow:
            # Get the source information
            source = await crud.source.get_by_short_name(
                db, short_name=source_connection_in.short_name
            )
            if not source:
                raise HTTPException(
                    status_code=404, detail=f"Source not found: {source_connection_in.short_name}"
                )

            # 1. Get or create integration credential
            integration_credential_id = None

            # If credential_id is provided, use it
            if aux_attrs.get("credential_id"):
                # Verify the credential exists and belongs to this user
                credential = await crud.integration_credential.get(
                    uow.session, id=aux_attrs["credential_id"], current_user=current_user
                )
                if not credential:
                    raise HTTPException(status_code=404, detail="Integration credential not found")

                if (
                    credential.integration_short_name != source_connection_in.short_name
                    or credential.integration_type != IntegrationType.SOURCE
                ):
                    raise HTTPException(
                        status_code=400, detail="Credential doesn't match the source type"
                    )

                integration_credential_id = credential.id
            # If auth_fields are provided, create new credential
            elif aux_attrs["auth_fields"] is not None:
                # Validate auth fields
                auth_fields = await self._validate_auth_fields(
                    db, source_connection_in.short_name, aux_attrs["auth_fields"]
                )

                # Create the integration credential
                integration_cred_in = schemas.IntegrationCredentialCreateEncrypted(
                    name=f"{source.name} - {current_user.email}",
                    description=f"Credentials for {source.name} - {current_user.email}",
                    integration_short_name=source_connection_in.short_name,
                    integration_type=IntegrationType.SOURCE,
                    auth_type=source.auth_type,
                    encrypted_credentials=credentials.encrypt(auth_fields),
                    auth_config_class=source.auth_config_class,
                )

                integration_credential = await crud.integration_credential.create(
                    uow.session, obj_in=integration_cred_in, current_user=current_user, uow=uow
                )

                await uow.session.flush()
                integration_credential_id = integration_credential.id
            else:
                # Neither credential_id nor auth_fields provided
                raise HTTPException(
                    status_code=400,
                    detail="Either auth_fields or credential_id must be "
                    "provided to create a source connection",
                )

            # Validate config fields (config_fields are in core_attrs)
            config_fields = await self._validate_config_fields(
                db, source_connection_in.short_name, core_attrs.get("config_fields")
            )

            # Always set the validated config_fields (even if empty)
            core_attrs["config_fields"] = config_fields

            # 2. Create the connection object for source (system table)
            connection_create = schemas.ConnectionCreate(
                name=source_connection_in.name,
                integration_type=IntegrationType.SOURCE,
                integration_credential_id=integration_credential_id,
                status=ConnectionStatus.ACTIVE,
                short_name=source_connection_in.short_name,
            )

            connection = await crud.connection.create(
                db=uow.session, obj_in=connection_create, current_user=current_user, uow=uow
            )

            await uow.session.flush()
            connection_id = connection.id

            # 3. Check if we need to create a collection first
            if "collection" not in core_attrs:
                # Create a collection with the same name as the source connection

                collection_create = schemas.CollectionCreate(
                    name=f"Collection for {source_connection_in.name}",
                    description=f"Auto-generated collection for {source_connection_in.name}",
                )

                collection = await collection_service.create(
                    db=uow.session,
                    collection_in=collection_create,
                    current_user=current_user,
                    uow=uow,
                )
            else:
                readable_collection_id = core_attrs["collection"]
                if "collection" in core_attrs:
                    del core_attrs["collection"]
                collection = await crud.collection.get_by_readable_id(
                    db=uow.session, readable_id=readable_collection_id, current_user=current_user
                )
                if not collection:
                    raise HTTPException(
                        status_code=404, detail=f"Collection '{readable_collection_id}' not found"
                    )

            # 4. Create the sync
            sync_in = schemas.SyncCreate(
                name=f"Sync for {source_connection_in.name}",
                description=f"Auto-generated sync for {source_connection_in.name}",
                source_connection_id=connection_id,  # ID of connection system table
                embedding_model_connection_id=NATIVE_TEXT2VEC_UUID,
                destination_connection_ids=[NATIVE_QDRANT_UUID],
                cron_schedule=aux_attrs["cron_schedule"],
                status=SyncStatus.ACTIVE,
                run_immediately=aux_attrs["sync_immediately"],
            )

            # 5. Use the sync service to create the sync and automatically the DAG
            sync, sync_job = await sync_service.create_and_run_sync(
                db=uow.session, sync_in=sync_in, current_user=current_user, uow=uow
            )

            # 6. Create the source connection from core attributes
            source_connection_create = {
                **core_attrs,
                "connection_id": connection_id,
                "readable_collection_id": collection.readable_id,
                "sync_id": sync.id,
                "white_label_id": core_attrs.get(
                    "white_label_id"
                ),  # Include white_label_id if provided
            }

            source_connection = await crud.source_connection.create(
                db=uow.session, obj_in=source_connection_create, current_user=current_user, uow=uow
            )
            await uow.session.flush()

            # map to schemas and return
            source_connection = schemas.SourceConnection.from_orm_with_collection_mapping(
                source_connection
            )

            # Only validate sync_job if it exists (when sync_immediately=True)
            if sync_job is not None:
                sync_job = schemas.SyncJob.model_validate(sync_job, from_attributes=True)

                # Update the source connection status with sync job info
                source_connection.status = SourceConnectionStatus.IN_PROGRESS
                source_connection.latest_sync_job_status = sync_job.status
                source_connection.latest_sync_job_id = sync_job.id
                source_connection.latest_sync_job_started_at = sync_job.started_at
                source_connection.latest_sync_job_completed_at = sync_job.completed_at
            else:
                # No sync job created (sync_immediately=False)
                source_connection.status = SourceConnectionStatus.ACTIVE
                source_connection.latest_sync_job_status = None
                source_connection.latest_sync_job_id = None
                source_connection.latest_sync_job_started_at = None
                source_connection.latest_sync_job_completed_at = None

            # Hide auth fields by default
            source_connection.auth_fields = "********"

        return source_connection, sync_job
Create a new source connection with all related objects.

        This method:
        1. Creates a credential with auth fields if provided, or uses existing credential
        2. Creates the connection to the source (schemas.Connection)
        3. Creates a collection if not provided
        4. Creates a sync configuration and DAG
        5. Creates a sync job if immediate execution is requested
        6. Creates the source connection (schemas.SourceConnection)

        Args:
            db: The database session
            source_connection_in: The source connection to create
            current_user: The current user

        Returns:
            A tuple of (source_connection, sync_job)

        Raises:
            HTTPException: If the source is not found
        
backend/airweave/core/source_connection_service.py