schema

api.schema.Query

Source code in api/schema.py

@type
class Query:
	@field
	async def all_address(
		self: Self,
		info: Info,
		filter: AddressFilterInput,
		page_size: PositiveInt = 10,
		page_number: PositiveInt = 1,
	) -> list[AddressType]:
		"""
		Query all addresses from database or all plugins.

		Args:
				info (Info): Strawberry default value to get context information
						in this case we use 'db'
				filter (AddressFilterInput): Strawberry input dataclass,
						everything can be None (based on sqlmodel model)
				page_size (PositiveInt, optional): How many elements in each page.
						Defaults to 10.
				page_number (PositiveInt, optional): Number of the page. Defaults to 1.

		Returns:
				list[AddressType]: All addresses (db model converted to strawberry type)
						based on filter or empty list

		"""
		result = await get_address(
			info.context.session, filter, page_size, page_number
		)
		if result['provider'] != 'local' and result['data']:
			info.context.background_tasks.add_task(
				functions.insert_address, info.context.session, result['data'][0]
			)
			# insert log

		return list(
			map(
				AddressType.from_pydantic,
				result['data'],
			)
		)

all_address(info, filter, page_size=10, page_number=1) async

Query all addresses from database or all plugins.

Parameters:

Name	Type	Description	Default
info	Info	Strawberry default value to get context information in this case we use 'db'	required
filter	AddressFilterInput	Strawberry input dataclass, everything can be None (based on sqlmodel model)	required
page_size	PositiveInt	How many elements in each page. Defaults to 10.	10
page_number	PositiveInt	Number of the page. Defaults to 1.	1

Returns:

Type	Description
list[AddressType]	list[AddressType]: All addresses (db model converted to strawberry type) based on filter or empty list

Source code in api/schema.py

@field
async def all_address(
	self: Self,
	info: Info,
	filter: AddressFilterInput,
	page_size: PositiveInt = 10,
	page_number: PositiveInt = 1,
) -> list[AddressType]:
	"""
	Query all addresses from database or all plugins.

	Args:
			info (Info): Strawberry default value to get context information
					in this case we use 'db'
			filter (AddressFilterInput): Strawberry input dataclass,
					everything can be None (based on sqlmodel model)
			page_size (PositiveInt, optional): How many elements in each page.
					Defaults to 10.
			page_number (PositiveInt, optional): Number of the page. Defaults to 1.

	Returns:
			list[AddressType]: All addresses (db model converted to strawberry type)
					based on filter or empty list

	"""
	result = await get_address(
		info.context.session, filter, page_size, page_number
	)
	if result['provider'] != 'local' and result['data']:
		info.context.background_tasks.add_task(
			functions.insert_address, info.context.session, result['data'][0]
		)
		# insert log

	return list(
		map(
			AddressType.from_pydantic,
			result['data'],
		)
	)

api.schema.Mutation

Source code in api/schema.py

@type
class Mutation:
	@field
	async def create_address(
		self: Self, info: Info, address: AddressInsertInput
	) -> AddressType:
		"""
		Insert address and city if not exists on database.

		Args:
				info (Info): Strawberry default value to get context information
						in this case we use 'db'
				address (AddressInsertInput): Strict address class,
						almost all fields need to be passed

		Returns:
				AddressType: Address (db model converted to strawberry dataclass)

		"""
		return AddressType.from_pydantic(
			await insert_address(info.context.session, address)
		)

create_address(info, address) async

Insert address and city if not exists on database.

Parameters:

Name	Type	Description	Default
info	Info	Strawberry default value to get context information in this case we use 'db'	required
address	AddressInsertInput	Strict address class, almost all fields need to be passed	required

Returns:

Name	Type	Description
AddressType	AddressType	Address (db model converted to strawberry dataclass)

Source code in api/schema.py

@field
async def create_address(
	self: Self, info: Info, address: AddressInsertInput
) -> AddressType:
	"""
	Insert address and city if not exists on database.

	Args:
			info (Info): Strawberry default value to get context information
					in this case we use 'db'
			address (AddressInsertInput): Strict address class,
					almost all fields need to be passed

	Returns:
			AddressType: Address (db model converted to strawberry dataclass)

	"""
	return AddressType.from_pydantic(
		await insert_address(info.context.session, address)
	)

api.schema.CustomContext

Bases: BaseContext

Source code in api/schema.py

class CustomContext(BaseContext):
	def __init__(self: Self, session: AsyncSession):
		"""Generate context database session."""
		self.session = session

__init__(session)

Generate context database session.

Source code in api/schema.py

def __init__(self: Self, session: AsyncSession):
	"""Generate context database session."""
	self.session = session

api.schema.get_context(session) async

Create database session to use when needed.

Parameters:

Name	Type	Description	Default
session	Annotated[AsyncSession, Depends	get db session from get_session	required

Returns:

Name	Type	Description
CustomContext	CustomContext	class that contains db session

Source code in api/schema.py

async def get_context(
	session: Annotated[AsyncSession, Depends(get_session)],
) -> CustomContext:
	"""
	Create database session to use when needed.

	Args:
			session (Annotated[AsyncSession, Depends): get db session from get_session

	Returns:
			CustomContext: class that contains db session

	"""
	return CustomContext(session)

api.schema.schema = Schema(query=Query, mutation=Mutation) module-attribute

Create Graphql Router for fastapi and start graphql ide if env is DEV.

api.schema.graphql_app = GraphQLRouter[object, object](schema, context_getter=get_context, graphql_ide='graphiql' if settings.DEV else None) module-attribute