diff --git a/src/components/Connections/CreateConnectionForm.test.tsx b/src/components/Connections/CreateConnectionForm.test.tsx
index cda86456..16281fb2 100644
--- a/src/components/Connections/CreateConnectionForm.test.tsx
+++ b/src/components/Connections/CreateConnectionForm.test.tsx
@@ -210,7 +210,8 @@ describe('Connections Setup', () => {
// check normalization after sync checkbox
const normalizationCheckbox = screen.getByTestId('normalizationCheckbox')
- .firstChild?.firstChild?.firstChild;
+ .firstChild?.firstChild;
+
expect(normalizationCheckbox).not.toBeChecked();
await act(() => userEvent.click(normalizationCheckbox));
expect(normalizationCheckbox).toBeChecked();
diff --git a/src/components/Connections/CreateConnectionForm.tsx b/src/components/Connections/CreateConnectionForm.tsx
index edaea724..de965d09 100644
--- a/src/components/Connections/CreateConnectionForm.tsx
+++ b/src/components/Connections/CreateConnectionForm.tsx
@@ -8,6 +8,11 @@ import {
Switch,
Select,
MenuItem,
+ RadioGroup,
+ Radio,
+ FormControl,
+ FormLabel,
+ Grid,
} from '@mui/material';
import {
Table,
@@ -156,13 +161,17 @@ const CreateConnectionForm = ({
setShowForm(false);
};
+ const handleRadioChange = (event: any) => {
+ setNormalize(event.target.value === 'normalized');
+ };
+
// create/update a connection
const onSubmit = async (data: any) => {
const payload: any = {
name: data.name,
sourceId: data.sources.id,
streams: sourceStreams,
- normalize: normalize,
+ normalize,
};
if (data.destinationSchema) {
payload.destinationSchema = data.destinationSchema;
@@ -279,22 +288,54 @@ const CreateConnectionForm = ({
- setNormalize(event.target.checked)}
- />
- }
- label="Normalize after sync?"
- />
+
+ Select type
+
+
+
+ }
+ label="Normalized"
+ />
+
+
+
+ }
+ label="Raw"
+ />
+
+
+
+
{sourceStreams.length > 0 && (
diff --git a/src/components/Destinations/CreateDestinationForm.tsx b/src/components/Destinations/CreateDestinationForm.tsx
index e8f928e3..eec7e1a6 100644
--- a/src/components/Destinations/CreateDestinationForm.tsx
+++ b/src/components/Destinations/CreateDestinationForm.tsx
@@ -154,6 +154,10 @@ const CreateDestinationForm = ({
});
}
+ // Todo: need to find a better way to do this
+ result = result.map((res: any) => ({ ...res, order: res.order + 1 }));
+ result.sort((a: any, b: any) => a.order - b.order);
+
return result;
};
diff --git a/src/components/Flows/FlowCreate.tsx b/src/components/Flows/FlowCreate.tsx
index 6b8d6831..260bc408 100644
--- a/src/components/Flows/FlowCreate.tsx
+++ b/src/components/Flows/FlowCreate.tsx
@@ -19,6 +19,8 @@ import Input from '../UI/Input/Input';
interface FlowCreateInterface {
updateCrudVal: (...args: any) => any;
mutate: (...args: any) => any;
+ flowId?: string;
+ setSelectedFlow?: (args: string) => any;
}
type ApiResponseConnection = {
@@ -32,19 +34,27 @@ type DispConnection = {
};
type DeploymentDef = {
+ active: boolean;
name: string;
dbtTransform: string;
connectionBlocks: Array;
- cron: string;
+ cron: string | object;
};
-const FlowCreate = ({ updateCrudVal, mutate }: FlowCreateInterface) => {
+const FlowCreate = ({
+ flowId,
+ updateCrudVal,
+ mutate,
+ setSelectedFlow = () => {},
+}: FlowCreateInterface) => {
+ const isEditPage = flowId !== '' && flowId !== undefined;
const { data: session } = useSession();
const toastContext = useContext(GlobalContext);
const [connections, setConnections] = useState([]);
- const { register, handleSubmit, control } = useForm({
+ const { register, handleSubmit, control, setValue } = useForm({
defaultValues: {
+ active: true,
name: '',
dbtTransform: 'no',
connectionBlocks: [],
@@ -53,20 +63,54 @@ const FlowCreate = ({ updateCrudVal, mutate }: FlowCreateInterface) => {
});
const handleClickCancel = () => {
+ setSelectedFlow('');
updateCrudVal('index');
};
- const processCronExpression = (cron: string) => {
- switch (cron) {
- case 'daily':
- return '0 1 * * *';
- case 'weekly':
- return '0 1 * * 1';
- default: // daily is the default
- return '0 1 * * *';
+ const convertCronExpression = (input: string) => {
+ const cronMappings: any = {
+ daily: '0 1 * * *',
+ weekly: '0 1 * * 1',
+ };
+
+ if (input in cronMappings) {
+ return cronMappings[input];
}
+
+ const reverseCronMappings = Object.fromEntries(
+ Object.entries(cronMappings).map(([key, value]) => [value, key])
+ );
+
+ return reverseCronMappings[input] || '0 1 * * *';
};
+ useEffect(() => {
+ if (flowId) {
+ (async () => {
+ try {
+ const data: any = await httpGet(session, `prefect/flows/${flowId}`);
+ setValue('name', data.name);
+ setValue('active', data.isScheduleActive);
+ setValue(
+ 'connectionBlocks',
+ data.parameters.airbyte_blocks.map((data: any) => data.name)
+ );
+ setValue(
+ 'dbtTransform',
+ data.parameters.dbt_blocks.length > 0 ? 'yes' : 'no'
+ );
+ setValue('cron', {
+ id: convertCronExpression(data.cron),
+ label: convertCronExpression(data.cron),
+ });
+ } catch (err: any) {
+ console.error(err);
+ errorToast(err.message, [], toastContext);
+ }
+ })();
+ }
+ }, [flowId]);
+
useEffect(() => {
(async () => {
try {
@@ -90,25 +134,44 @@ const FlowCreate = ({ updateCrudVal, mutate }: FlowCreateInterface) => {
}, []);
const onSubmit = async (data: any) => {
- console.log(data);
try {
- const blocks = data.connectionBlocks.map((block: any, index: number) => ({
- ...block,
- seq: index + 1,
- }));
- const response = await httpPost(session, 'prefect/flows/', {
- name: data.name,
- connectionBlocks: blocks,
- dbtTransform: data.dbtTransform,
- cron: processCronExpression(data.cron.id),
- });
- mutate();
- updateCrudVal('index');
- successToast(
- `Flow ${response.name} created successfully`,
- [],
- toastContext
- );
+ if (isEditPage) {
+ await httpPost(
+ session,
+ `prefect/flows/${flowId}/set_schedule/${
+ data.active ? 'active' : 'inactive'
+ }`,
+ {}
+ );
+ successToast(
+ `Flow ${data.name} updated successfully`,
+ [],
+ toastContext
+ );
+ setSelectedFlow('');
+ mutate();
+ updateCrudVal('index');
+ } else {
+ const blocks = data.connectionBlocks.map(
+ (block: any, index: number) => ({
+ ...block,
+ seq: index + 1,
+ })
+ );
+ const response = await httpPost(session, 'prefect/flows/', {
+ name: data.name,
+ connectionBlocks: blocks,
+ dbtTransform: data.dbtTransform,
+ cron: convertCronExpression(data.cron.id),
+ });
+ mutate();
+ updateCrudVal('index');
+ successToast(
+ `Flow ${response.name} created successfully`,
+ [],
+ toastContext
+ );
+ }
} catch (err: any) {
console.error(err);
errorToast(err.message, [], toastContext);
@@ -124,7 +187,7 @@ const FlowCreate = ({ updateCrudVal, mutate }: FlowCreateInterface) => {
gutterBottom
color="#000"
>
- Create a new Flow
+ {flowId ? 'Update flow' : 'Create a new Flow'}
{
Flow details
+ {isEditPage && (
+
+ (
+
+ {
+ onChange(value);
+ }}
+ />
+
+ )}
+ />
+
+ Is Active ?
+
+
+ )}
{
render={({ field }: any) => (
{
render={({ field: { value, onChange } }) => (
{
onChange(value ? 'yes' : 'no');
@@ -245,6 +334,8 @@ const FlowCreate = ({ updateCrudVal, mutate }: FlowCreateInterface) => {
render={({ field }) => (
;
updateCrudVal: (...args: any) => any;
mutate: (...args: any) => any;
+ setSelectedFlow: (arg: string) => any;
}
const flowState = (flow: FlowInterface) => {
@@ -98,7 +99,12 @@ const flowLastRun = (flow: FlowInterface) => {
);
};
-export const Flows = ({ flows, updateCrudVal, mutate }: FlowsInterface) => {
+export const Flows = ({
+ flows,
+ updateCrudVal,
+ mutate,
+ setSelectedFlow,
+}: FlowsInterface) => {
const [showFlowRunHistory, setShowFlowRunHistory] = useState(false);
const [flowRunHistoryDeploymentId, setFlowRunHistoryDeploymentId] =
useState('');
@@ -120,6 +126,12 @@ export const Flows = ({ flows, updateCrudVal, mutate }: FlowsInterface) => {
setShowConfirmDeleteDialog(true);
};
+ const handleEditConnection = () => {
+ handleClose();
+ setSelectedFlow(deploymentId);
+ updateCrudVal('update');
+ };
+
const handleClick = (blockId: string, event: HTMLElement | null) => {
setDeploymentId(blockId);
setAnchorEl(event);
@@ -143,7 +155,7 @@ export const Flows = ({ flows, updateCrudVal, mutate }: FlowsInterface) => {
color="rgba(9, 37, 64, 0.87)"
fontWeight={700}
>
- by 12pm time every day
+ by 12pm every day
,
flowStatus(flow.status),
@@ -253,6 +265,7 @@ export const Flows = ({ flows, updateCrudVal, mutate }: FlowsInterface) => {
eleType="flow"
anchorEl={anchorEl}
open={open}
+ handleEdit={handleEditConnection}
handleClose={handleClose}
elementId={deploymentId}
handleDeleteConnection={handleDeleteConnection}
diff --git a/src/components/UI/Menu/Menu.tsx b/src/components/UI/Menu/Menu.tsx
index 50ffa741..dde342a7 100644
--- a/src/components/UI/Menu/Menu.tsx
+++ b/src/components/UI/Menu/Menu.tsx
@@ -1,6 +1,7 @@
import { Divider, ListItemIcon, Menu, MenuItem } from '@mui/material';
import Image from 'next/image';
import EditIcon from '@/assets/icons/edit.svg';
+import RestartAltIcon from '@mui/icons-material/RestartAlt';
import DeleteIcon from '@/assets/icons/delete.svg';
interface MenuProps {
@@ -59,7 +60,7 @@ export const ActionsMenu: React.FC = ({
{eleType === 'connection' && handleResetConnection && (
diff --git a/src/config/theme.ts b/src/config/theme.ts
index b8077bcc..9d3ce413 100644
--- a/src/config/theme.ts
+++ b/src/config/theme.ts
@@ -47,6 +47,22 @@ const theme = createTheme({
},
},
},
+ MuiFormLabel: {
+ styleOverrides: {
+ root: {
+ fontWeight: 600,
+ marginBottom: 10,
+ color: '#758397',
+ },
+ },
+ },
+ MuiFormControlLabel: {
+ styleOverrides: {
+ root: {
+ color: '#758397',
+ },
+ },
+ },
MuiTextField: {
styleOverrides: {
root: {
diff --git a/src/pages/pipeline/orchestrate.tsx b/src/pages/pipeline/orchestrate.tsx
index 7afd23e2..ad87b85b 100644
--- a/src/pages/pipeline/orchestrate.tsx
+++ b/src/pages/pipeline/orchestrate.tsx
@@ -9,6 +9,7 @@ import { backendUrl } from '@/config/constant';
export default function Orchestrate() {
const [crudVal, setCrudVal] = useState('index'); // can be index or create
const [flows, setFlows] = useState>([]);
+ const [selectedFlow, setSelectedFlow] = useState('');
const updateCrudVal = (crudState: string) => {
setCrudVal(crudState);
@@ -27,11 +28,25 @@ export default function Orchestrate() {
<>
- {crudVal === 'index' ? (
-
- ) : (
+ {crudVal === 'index' && (
+
+ )}
+ {crudVal === 'create' && (
)}
+ {crudVal === 'update' && (
+
+ )}
>
);