-
Notifications
You must be signed in to change notification settings - Fork 1
Expand file tree
/
Copy pathmain.py
More file actions
86 lines (66 loc) · 2.13 KB
/
main.py
File metadata and controls
86 lines (66 loc) · 2.13 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
"""
Main module for the Clubs Microservice.
This module sets up the FastAPI application and integrates the Strawberry
GraphQL schema.
It includes the configuration for queries, mutations, and context.
Attributes:
GLOBAL_DEBUG (str): Environment variable that Enables or Disables debug
mode. Defaults to "False".
DEBUG (bool): Indicates whether the application is running in debug mode.
gql_app (GraphQLRouter): The GraphQL router for handling GraphQL requests.
app (FastAPI): The FastAPI application instance.
"""
from contextlib import asynccontextmanager
from os import getenv
import strawberry
from fastapi import FastAPI
from strawberry.extensions import DisableIntrospection, PydanticErrorExtension
from strawberry.fastapi import GraphQLRouter
from strawberry.tools import create_type
# override PyObjectId and Context scalars
from db import ensure_clubs_index
from models import PyObjectId
from mutations import mutations
from otypes import Context, PyObjectIdType
# import all queries and mutations
from queries import queries
# create query types
Query = create_type("Query", queries)
# create mutation types
Mutation = create_type("Mutation", mutations)
# Strawberry extensions
extensions = [
PydanticErrorExtension,
]
# Returns The custom context by overriding the context getter.
def get_context() -> Context:
return Context()
# check whether running in debug mode
DEBUG = getenv("GLOBAL_DEBUG", "False").lower() in ("true", "1", "t")
if DEBUG:
print("Running in debug mode")
else:
extensions += [
DisableIntrospection(),
]
schema = strawberry.federation.Schema(
query=Query,
mutation=Mutation,
scalar_overrides={PyObjectId: PyObjectIdType},
extensions=extensions,
)
# serve API with FastAPI router
gql_app = GraphQLRouter(schema, context_getter=get_context)
@asynccontextmanager
async def lifespan(app: FastAPI):
# Startup
await ensure_clubs_index()
yield
# Shutdown
app = FastAPI(
debug=DEBUG,
title="CC Clubs Microservice",
desciption="Handles Data of Clubs",
lifespan=lifespan,
)
app.include_router(gql_app, prefix="/graphql")