Repository: prefecthq/prefect
Issue: Parameter 'priority' is ignored when creating work queues

### First check

- [X] I added a descriptive title to this issue.
- [X] I used the GitHub search to find a similar issue and didn't find it.
- [X] I searched the Prefect documentation for this issue.
- [X] I checked that this issue is related to Prefect and not one of its dependencies.

### Bug summary

I wrote a function to create 3 work queues with specified priority (10, 20 and 30). Work queues are created but priority is the default one (2,3,4).

![Screenshot from 2023-05-31 13-46-29](https://github.com/PrefectHQ/prefect/assets/4247916/2600f273-cffe-428f-8cd6-6653c913b88b)


### Reproduction

```python3
async def create_work_queues():
    async with get_client() as client:
        print("Creating work queues ...")
        queues = await client.read_work_queues("default-agent-pool")
        names = {queue.name for queue in queues}
        print(f"Existing work queues: {names}")
        tasks = []
        if "high" not in names:
            tasks.append(client.create_work_queue("high", priority=10))
        if "medium" not in names:
            tasks.append(client.create_work_queue("medium", priority=20))
        if "low" not in names:
            tasks.append(client.create_work_queue("low", priority=30))

        await asyncio.gather(*tasks)
        print("Finished creating work queues")
```


### Error

_No response_

### Versions

```Text
Version:             2.10.11
API version:         0.8.4
Python version:      3.11.3
Git commit:          8c651ffc
Built:               Thu, May 25, 2023 2:59 PM
OS/Arch:             linux/x86_64
Profile:             default
Server type:         server
```


### Additional context

_No response_

Changed files:
src/prefect/server/models/work_queues.py, tests/server/api/test_work_queues.py

diff --git a/src/prefect/server/models/work_queues.py b/src/prefect/server/models/work_queues.py
index c60484a07..bc6ee0f15 100644
--- a/src/prefect/server/models/work_queues.py
+++ b/src/prefect/server/models/work_queues.py
@@ -64,14 +64,17 @@ async def create_work_queue(
                 )
             data["work_pool_id"] = default_agent_work_pool.id
 
-    # Set the priority to be the max priority + 1
-    # This will make the new queue the lowest priority
-    max_priority_query = sa.select(
-        sa.func.coalesce(sa.func.max(db.WorkQueue.priority), 0)
-    ).where(db.WorkQueue.work_pool_id == data["work_pool_id"])
-    priority = (await session.execute(max_priority_query)).scalar()
-
-    model = db.WorkQueue(**data, priority=priority + 1)
+    if work_queue.priority is not None:
+        priority = work_queue.priority
+    else:
+        # Set the default priority to be the max priority + 1
+        # This will make the new queue the lowest priority
+        max_priority_query = sa.select(
+            sa.func.coalesce(sa.func.max(db.WorkQueue.priority), 0)
+        ).where(db.WorkQueue.work_pool_id == data["work_pool_id"])
+        priority = (await session.execute(max_priority_query)).scalar() + 1
+
+    model = db.WorkQueue(**data, priority=priority)
 
     session.add(model)
     await session.flush()
diff --git a/tests/server/api/test_work_queues.py b/tests/server/api/test_work_queues.py
index 8e688120a..eb96bea43 100644
--- a/tests/server/api/test_work_queues.py
+++ b/tests/server/api/test_work_queues.py
@@ -54,6 +54,87 @@ class TestCreateWorkQueue:
         assert response.status_code == status.HTTP_422_UNPROCESSABLE_ENTITY
         assert b"contains an invalid character" in response.content
 
+    async def test_create_work_queue_with_priority(
+        self,
+        session,
+        client,
+    ):
+        data = WorkQueueCreate(name="wq-1", priority=5).dict(json_compatible=True)
+        response = await client.post("/work_queues/", json=data)
+        assert response.status_code == status.HTTP_201_CREATED
+        assert response.json()["name"] == "wq-1"
+        assert response.json()["priority"] == 5
+        work_queue_id = response.json()["id"]
+
+        work_queue = await models.work_queues.read_work_queue(
+            session=session, work_queue_id=work_queue_id
+        )
+        assert str(work_queue.id) == work_queue_id
+        assert work_queue.name == "wq-1"
+        assert work_queue.priority == 5
+
+    async def test_create_work_queue_with_default_priority(
+        self,
+        session,
+        client,
+    ):
+        data = WorkQueueCreate(name="wq-1").dict(json_compatible=True)
+        response = await client.post("/work_queues/", json=data)
+        assert response.status_code == status.HTTP_201_CREATED
+
+        await models.work_queues.read_work_queues(session=session)
+        assert response.json()["name"] == "wq-1"
+        # Starts at 2 because the 'default' work queue is 1
+        assert response.json()["priority"] == 2
+        work_queue_id = response.json()["id"]
+
+        work_queue = await models.work_queues.read_work_queue(
+            session=session, work_queue_id=work_queue_id
+        )
+        assert str(work_queue.id) == work_queue_id
+        assert work_queue.name == "wq-1"
+        assert work_queue.priority == 2
+
+        data = WorkQueueCreate(name="wq-2").dict(json_compatible=True)
+        response = await client.post("/work_queues/", json=data)
+        assert response.status_code == status.HTTP_201_CREATED
+        assert response.json()["name"] == "wq-2"
+        assert response.json()["priority"] == 3
+        work_queue_id = response.json()["id"]
+
+        work_queue = await models.work_queues.read_work_queue(
+            session=session, work_queue_id=work_queue_id
+        )
+        assert str(work_queue.id) == work_queue_id
+        assert work_queue.name == "wq-2"
+        assert work_queue.priority == 3
+
+    async def test_create_work_queue_with_existing_priority(
+        self,
+        session,
+        client,
+    ):
+        data = WorkQueueCreate(name="wq-1", priority=5).dict(json_compatible=True)
+        response = await client.post("/work_queues/", json=data)
+        assert response.status_code == status.HTTP_201_CREATED
+        assert response.json()["name"] == "wq-1"
+        assert response.json()["priority"] == 5
+        work_queue_id = response.json()["id"]
+
+        data = WorkQueueCreate(name="wq-2", priority=5).dict(json_compatible=True)
+        response = await client.post("/work_queues/", json=data)
+        assert response.status_code == status.HTTP_201_CREATED
+        assert response.json()["name"] == "wq-2"
+        assert response.json()["priority"] == 5
+        work_queue_id = response.json()["id"]
+
+        work_queue = await models.work_queues.read_work_queue(
+            session=session, work_queue_id=work_queue_id
+        )
+        assert str(work_queue.id) == work_queue_id
+        assert work_queue.name == "wq-2"
+        assert work_queue.priority == 5
+
 
 class TestUpdateWorkQueue:
     async def test_update_work_queue(