Add type hints and mypy
All checks were successful
continuous-integration/drone/push Build is passing

This commit is contained in:
Paul Brinkmeier 2023-01-19 03:42:25 +01:00
parent 9cd3803b61
commit a2a17e5ff9
6 changed files with 58 additions and 21 deletions

View File

@ -1,11 +1,23 @@
---
kind: pipeline
type: docker
name: client
name: frontend
steps:
- name: build
image: node:19.4-alpine
commands:
- cd glebby-client
- npm install
- npm run build
- VITE_GLEBBY_SERVER_URL=ws://localhost:5000/glebby npx vite build
---
kind: pipeline
type: docker
name: backend
steps:
- name: typecheck
image: python:3.11-alpine
commands:
- cd glebby-server
- pip install -r requirements-dev.txt
- mypy --strict glebby

View File

@ -4,3 +4,4 @@
- [ ] lobby
- [ ] actual game logic
- [ ] websocket url config
- [ ] use poetry?

View File

@ -1,28 +1,28 @@
import asyncio
from quart import Quart, websocket
from quart import Quart, Response, websocket
from glebby.model import Model
model = Model()
app = Quart(__name__, static_url_path='')
async def send_outgoing(queue):
async def send_outgoing(queue: asyncio.Queue[str]) -> None:
while True:
data = await queue.get()
await websocket.send(data)
@app.route('/')
async def root():
async def root() -> Response:
return await app.send_static_file('index.html')
@app.websocket('/glebby')
async def ws():
async def ws() -> None:
global model
# For each connected client, we create a queue to store its outgoing messages.
# We pass this queue to the model so that it can communicate with the clients.
outgoing_queue = asyncio.Queue()
outgoing_queue: asyncio.Queue[str] = asyncio.Queue()
client_id = await model.add_client(outgoing_queue)
try:

View File

@ -1,7 +1,11 @@
from random import Random
from typing import Optional
class BoardGenerator:
def __init__(self, seed):
dice: list[str]
rng: Random
def __init__(self, seed: Optional[int]):
self.dice = [
'aeaneg',
'wngeeh',
@ -22,7 +26,7 @@ class BoardGenerator:
]
self.rng = Random(seed)
def generate_board(self):
def generate_board(self) -> list[list[str]]:
shuffled_dice = self.rng.sample(self.dice, k=len(self.dice))
roll_results = [self.rng.choice(die) for die in shuffled_dice]
return [roll_results[4 * i : 4 * (i+1)] for i in range(0, 4)]

View File

@ -1,26 +1,41 @@
import asyncio
import itertools
import json
from collections import OrderedDict
from typing import Any, Iterator, Optional
from glebby.board import BoardGenerator
class Client:
def __init__(self, client_id, outgoing_queue):
outgoing_queue: asyncio.Queue[str]
id: int
name: str
def __init__(self, client_id: int, outgoing_queue: asyncio.Queue[str]):
self.outgoing_queue = outgoing_queue
self.data = { 'id': client_id, 'name': '' }
self.id = client_id
self.name = ''
async def put_outgoing_message(self, message):
async def put_outgoing_message(self, message: str) -> None:
await self.outgoing_queue.put(message)
def get_dto(self) -> dict[str, Any]:
return { 'id': self.id, 'name': self.id }
class Model:
def __init__(self):
clients: OrderedDict[int, Client]
id_generator: Iterator[int]
board_generator: BoardGenerator
board: Optional[list[list[str]]]
def __init__(self) -> None:
self.clients = OrderedDict()
self.id_generator = itertools.count(start=0, step=1)
self.board_generator = BoardGenerator(None)
self.board = None
async def add_client(self, outgoing_queue):
async def add_client(self, outgoing_queue: asyncio.Queue[str]) -> int:
client_id = next(self.id_generator)
await self.broadcast(client_id, { 'type': 'join' })
@ -34,28 +49,28 @@ class Model:
return client_id
async def remove_client(self, client_id):
async def remove_client(self, client_id: int) -> None:
del self.clients[client_id]
await self.broadcast(client_id, { 'type': 'leave' })
print(f'<{client_id}|leave>')
async def send_to(self, client_id_from, client_id_to, payload):
async def send_to(self, client_id_from: Optional[int], client_id_to: int, payload: Any) -> None:
await self.clients[client_id_to].put_outgoing_message(json.dumps({
'from': client_id_from,
'payload': payload
}))
async def broadcast(self, client_id_from, payload):
async def broadcast(self, client_id_from: Optional[int], payload: Any) -> None:
for client_id_to in self.clients:
await self.send_to(client_id_from, client_id_to, payload)
async def handle_incoming(self, client_id, message):
async def handle_incoming(self, client_id: int, message: str) -> None:
print(f'<{client_id}|data> {message}')
payload = json.loads(message)
if payload['type'] == 'set-name':
self.clients[client_id].data['name'] = payload['name']
self.clients[client_id].name = payload['name']
await self.broadcast(client_id, {
'type': 'set-name',
'name': payload['name']
@ -74,11 +89,11 @@ class Model:
else:
print("Unhandled!")
def get_state_dto(self, client_id):
def get_state_dto(self, client_id: int) -> Any:
return {
'yourId': client_id,
'players': [
client.data
client.get_dto()
for client in self.clients.values()
],
'board': self.board

View File

@ -0,0 +1,5 @@
-r requirements.txt
mypy==0.991
mypy-extensions==0.4.3
tomli==2.0.1
typing_extensions==4.4.0