Merge branch 'master' into hasura-metadata

.circleci/config.yml

@@ -191,7 +191,7 @@ workflows:
           docker-username: DOCKER_HUB_USER
           image: 'prefecthq/apollo'
           path: 'services/apollo'
-          extra_build_args: '--build-arg PREFECT_VERSION=master'
+          extra_build_args: '--build-arg PREFECT_VERSION=master --build-arg RELEASE_TIMESTAMP=$(date -u +"%Y-%m-%dT%H:%M:%SZ")'
           tag: master
           filters:
             branches:
@@ -215,6 +215,7 @@ workflows:
           docker-username: DOCKER_HUB_USER
           image: 'prefecthq/apollo'
           path: 'services/apollo'
+          extra_build_args: '--build-arg PREFECT_VERSION=master --build-arg RELEASE_TIMESTAMP=$(date -u +"%Y-%m-%dT%H:%M:%SZ")'
           tag: $CIRCLE_TAG,latest
           filters:
             branches:

Changelog.md

@@ -1,5 +1,29 @@
 # Changelog
 
+## September 11, 2020 <Badge text="beta" type="success" />
+
+Released on September 11, 2020.
+
+### Features
+
+- Add database structure and routes for persisting Agents - [#58](https://github.com/PrefectHQ/server/pull/58)
+
+### Enhancements
+
+- Add route for querying mapped children - [#73](https://github.com/PrefectHQ/server/pull/73)
+
+### Fixes
+
+- Fix __init__.py for deprecated API module - [#71](https://github.com/PrefectHQ/server/pull/71)
+
+### Breaking Changes
+
+- Remove `run_count` and `duration` columns - [#72](https://github.com/PrefectHQ/server/pull/72)
+
+### Database Migrations
+
+- Increase performance of state update triggers - [#72](https://github.com/PrefectHQ/server/pull/72)
+
 ## September 1, 2020 <Badge text="beta" type="success" />
 
 Released on September 1, 2020.

changes/issue86.yaml

@@ -0,0 +1,2 @@
+fix:
+  - "Fix and consolidate behavior for active schedules - [#86](https://github.com/PrefectHQ/server/issues/86)"

changes/pr76.yml

@@ -0,0 +1,5 @@
+enhancement:
+  - "Reduce size of `prefecthq/apollo` Docker image - [#3215](https://github.com/PrefectHQ/server/pull/76)"
+
+contributor:
+  - "[Juan Calderon-Perez](https://github.com/gabrielcalderon)"

changes/pr89.yml

@@ -0,0 +1,5 @@
+fix:
+  - "Fix agent API calling `prefect.api.models` instead of `prefect.models` - [#89](https://github.com/PrefectHQ/server/pull/89)"
+
+enhancement:
+  - "Replace calls to `prefect_server.database.models` with `prefect.models` - [#89](https://github.com/PrefectHQ/server/pull/89)"

services/apollo/.dockerignore

@@ -0,0 +1,2 @@
+node_modules
+npm-debug.log

services/apollo/Dockerfile

@@ -1,15 +1,30 @@
-FROM node:10.15.0
-ARG VERSION
-ENV PREFECT_SERVER_VERSION=${VERSION}
+ARG NODE_VERSION=${NODE_VERSION:-10.15.0}
+FROM node:${NODE_VERSION}-slim
 
-COPY . /apollo
+# Prefect Version, default to MASTER
+ARG PREFECT_VERSION
+ENV PREFECT_VERSION=${PREFECT_VERSION:-master}
 
-RUN \
-  cd /apollo \
-  && npm ci && npm run build
+# Prefect Server Version, default to MASTER
+ARG PREFECT_SERVER_VERSION
+ENV PREFECT_SERVER_VERSION=${PREFECT_SERVER_VERSION:-master}
+
+ARG RELEASE_TIMESTAMP
+ENV RELEASE_TIMESTAMP=$RELEASE_TIMESTAMP
+
+# Image Labels
+LABEL maintainer="[email protected]"
+LABEL org.label-schema.schema-version = "1.0"
+LABEL org.label-schema.name="apollo"
+LABEL org.label-schema.url="https://www.prefect.io/"
+LABEL org.label-schema.version=${PREFECT_VERSION}
+LABEL org.label-schema.build-date=${RELEASE_TIMESTAMP}
 
 WORKDIR /apollo
-COPY post-start.sh .
-RUN chmod +x post-start.sh
+COPY . .
+
+RUN npm ci && \
+    npm run build && \
+    chmod +x post-start.sh
 
 CMD ["npm", "run", "serve"]

setup.py

@@ -6,8 +6,9 @@
 
 install_requires = [
     "prefect >= 0.13.0",
-    "ariadne >= 0.8.0, < 0.12.0",
     "alembic >= 1.2, < 2.0",
+    "ariadne >= 0.8.0, < 0.12.0",
+    "asyncpg >= 0.20, < 0.21",
     "click >= 6.7, <8.0",
     "coolname >= 1.1, < 2.0",
     "docker >= 3.4,< 5.0",

src/prefect_server/api/agents.py

@@ -2,9 +2,8 @@
 
 import pendulum
 
-from prefect_server.database import models
+from prefect import models
 from prefect_server.utilities import context
-from prefect import api
 from prefect.utilities.plugins import register_api
 
 
@@ -37,7 +36,7 @@ async def register_agent(
             tenant_id = await models.Tenant.where({"id": {"_eq": None}}).first().id
         except:
             raise ValueError("No tenant found.")
-    return await api.models.Agent(
+    return await models.Agent(
         tenant_id=tenant_id,
         agent_config_id=agent_config_id,
         labels=labels or [],
@@ -60,7 +59,7 @@ async def update_agent_last_queried(agent_id: str) -> bool:
     """
     if agent_id is None:
         raise ValueError("Must supply an agent ID to update.")
-    result = await api.models.Agent.where(id=agent_id).update(
+    result = await models.Agent.where(id=agent_id).update(
         set={"last_queried": pendulum.now("utc")}
     )
     return bool(result.affected_rows)  # type: ignore
@@ -79,7 +78,7 @@ async def delete_agent(agent_id: str) -> bool:
     """
     if agent_id is None:
         raise ValueError("Must supply an agent ID to delete.")
-    result = await api.models.Agent.where(id=agent_id).delete()
+    result = await models.Agent.where(id=agent_id).delete()
     return bool(result.affected_rows)  # type: ignore
 
 
@@ -100,7 +99,7 @@ async def create_agent_config(
     Returns:
         - str: the agent config id
     """
-    return await api.models.AgentConfig(
+    return await models.AgentConfig(
         tenant_id=tenant_id, name=name, settings=settings
     ).insert()
 
@@ -118,7 +117,7 @@ async def delete_agent_config(agent_config_id: str) -> bool:
     """
     if agent_config_id is None:
         raise ValueError("Must supply an agent ID to delete.")
-    result = await api.models.AgentConfig.where(id=agent_config_id).delete()
+    result = await models.AgentConfig.where(id=agent_config_id).delete()
     return bool(result.affected_rows)  # type: ignore
 
 
@@ -146,5 +145,5 @@ async def update_agent_config(
     if settings is not None:
         update["settings"] = settings
 
-    result = await api.models.AgentConfig.where(id=agent_config_id).update(set=update)
+    result = await models.AgentConfig.where(id=agent_config_id).update(set=update)
     return bool(result.affected_rows)  # type: ignore

src/prefect_server/api/cloud_hooks.py

@@ -8,9 +8,8 @@
 from pydantic import BaseModel
 
 import prefect
-from prefect import api
+from prefect import api, models
 from prefect_server import config as server_config
-from prefect_server.database import models
 from prefect_server.utilities import logging, names, events
 from prefect.utilities.plugins import register_api
 

src/prefect_server/api/flow_groups.py

@@ -2,8 +2,7 @@
 
 from prefect.serialization.schedule import ClockSchema
 
-from prefect import api
-from prefect_server.database import models
+from prefect import api, models
 from prefect.utilities.plugins import register_api
 
 

src/prefect_server/api/flows.py

@@ -8,9 +8,8 @@
 
 from prefect.serialization.schedule import ScheduleSchema
 from prefect.utilities.graphql import with_args
-from prefect import api
+from prefect import api, models
 from prefect_server import config
-from prefect_server.database import models
 from prefect_server.utilities import logging
 from prefect.utilities.plugins import register_api
 
@@ -153,17 +152,6 @@ async def create_flow(
         raise ValueError("Invalid project.")
     tenant_id = project.tenant_id  # type: ignore
 
-    # check required parameters - can't load a flow that has required params and a shcedule
-    # NOTE: if we allow schedules to be set via UI in the future, we might skip or
-    # refactor this check
-    required_parameters = [p for p in flow.parameters if p.required]
-    if flow.schedule is not None and required_parameters:
-        required_names = {p.name for p in required_parameters}
-        if not all(
-            [required_names <= set(c.parameter_defaults) for c in flow.schedule.clocks]
-        ):
-            raise ValueError("Can not schedule a flow that has required parameters.")
-
     # set up task detail info
     task_lookup = {t.slug: t for t in flow.tasks}
     tasks_with_upstreams = {e.downstream_task for e in flow.edges}
@@ -229,7 +217,7 @@ async def create_flow(
         flow_group_id=flow_group_id,
         description=description,
         schedule=serialized_flow.get("schedule"),
-        is_schedule_active=set_schedule_active,
+        is_schedule_active=False,
         tasks=[
             models.Task(
                 id=t.id,
@@ -264,7 +252,12 @@ async def create_flow(
 
     # schedule runs
     if set_schedule_active:
-        await api.flows.schedule_flow_runs(flow_id=flow_id)
+        # we don't want to error the Flow creation call as it would prevent other archiving logic
+        # from kicking in
+        try:
+            await api.flows.set_schedule_active(flow_id=flow_id)
+        except ValueError:
+            pass
 
     return flow_id
 
@@ -397,6 +390,34 @@ async def set_schedule_active(flow_id: str) -> bool:
     if flow_id is None:
         raise ValueError("Invalid flow id.")
 
+    flow = await models.Flow.where(id=flow_id).first(
+        {
+            "schedule": True,
+            "parameters": True,
+            "flow_group": {"schedule": True, "default_parameters": True},
+        }
+    )
+
+    if not flow:
+        return False
+
+    # logic for determining if it's appropriate to turn on the schedule for this flow
+    # we can set a flow run schedule to active if any required parameters are provided by:
+    # - the Flow's own clocks
+    # - the Flow Group's default parameters
+    # - some combination of the above two
+    required_parameters = {p.get("name") for p in flow.parameters if p.get("required")}
+    if flow.schedule is not None and required_parameters:
+        required_names = required_parameters.difference(
+            flow.flow_group.default_parameters or {}
+        )
+        clock_params = [
+            set(c.get("parameter_defaults", {}).keys())
+            for c in flow.schedule.get("clocks", [])
+        ]
+        if not all([required_names <= c for c in clock_params]):
+            raise ValueError("Can not schedule a flow that has required parameters.")
+
     result = await models.Flow.where(id=flow_id).update(
         set={"is_schedule_active": True}
     )

src/prefect_server/api/logs.py

@@ -4,7 +4,7 @@
 import pendulum
 import pydantic
 
-from prefect_server.database import models
+from prefect import models
 from prefect.utilities.plugins import register_api
 
 

src/prefect_server/api/messages.py

@@ -1,4 +1,4 @@
-from prefect_server.database import models
+from prefect import models
 from prefect.utilities.plugins import register_api
 
 PREFECT_MESSAGE_TYPES = {"CLOUD_HOOK", "REQUIRES_APPROVAL"}

src/prefect_server/api/projects.py

@@ -1,4 +1,4 @@
-from prefect_server.database import models
+from prefect import models
 from prefect.utilities.plugins import register_api
 
 

src/prefect_server/api/runs.py

@@ -6,9 +6,8 @@
 import prefect
 from prefect.engine.state import Pending, Queued, Scheduled
 from prefect.utilities.graphql import EnumValue
-from prefect import api
+from prefect import api, models
 from prefect_server import config
-from prefect_server.database import models
 from prefect_server.utilities import exceptions, names
 from prefect.utilities.plugins import register_api
 

src/prefect_server/api/states.py

@@ -9,8 +9,7 @@
 
 import prefect
 from prefect.engine.state import Cancelled, Cancelling, State
-from prefect import api
-from prefect_server.database import models
+from prefect import api, models
 from prefect_server.utilities import events
 from prefect_server.utilities.logging import get_logger
 from prefect.utilities.plugins import register_api

src/prefect_server/api/tenants.py

@@ -1,6 +1,6 @@
 import slugify
 
-from prefect_server.database import models
+from prefect import models
 from prefect.utilities.plugins import register_api
 
 

src/prefect_server/database/__init__.py

@@ -1,3 +1,4 @@
 import prefect_server.database.hasura
 import prefect_server.database.orm
 import prefect_server.database.models
+import prefect_server.database.postgres