### Demonstrate Plugin Execution

Source: https://github.com/aiokitchen/aiomisc/blob/master/docs/source/plugins.md

This example shows how to manually call the plugin's setup function and run the aiomisc entrypoint to trigger the plugin.

```python
# Content of: ``my_plugin_example.py``
# ======================================================================
# The code below is not related to the plugin, but serves to demonstrate
# how it works.
# ======================================================================
def main():
    """ some function in user code """

    # This function will be called by aiomisc.plugin module
    # in this example it's just for demonstration.
    setup()

    assert not event.is_set()

    with aiomisc.entrypoint() as loop:
        pass

    assert event.is_set()

main()

# remove the plugin on when unneeded
aiomisc.entrypoint.PRE_START.disconnect(hello)
```

--------------------------------

### Example Plugin Module Structure

Source: https://github.com/aiokitchen/aiomisc/blob/master/docs/source/plugins.md

A plugin module must have a setup() function that connects signals. The 'hello' function demonstrates signal handling.

```python
# Content of: ``aiomisc_myplugin/plugin.py``
from typing import Tuple
from threading import Event

import aiomisc


event = Event()
# Will be shown in ``python -m aiomisc.plugins``
__doc__ = "Example plugin"


async def hello(
    *,
    entrypoint: aiomisc.Entrypoint,
    services: Tuple[aiomisc.Service, ...]
) -> None:
    print('Hello from aiomisc plugin')
    event.set()


def setup() -> None:
    """
    This code will be called by loading plugins declared in
    ``pyproject.toml`` or ``setup.py``.
    """
    aiomisc.Entrypoint.PRE_START.connect(hello)
```

--------------------------------

### Install aiomisc with multiple extras

Source: https://github.com/aiokitchen/aiomisc/blob/master/docs/source/index.md

Install aiomisc with multiple optional dependencies by separating them with commas. This example includes aiohttp, cron, rich, and uvloop.

```bash
pip3 install "aiomisc[aiohttp,cron,rich,uvloop]"
```

--------------------------------

### Install Pillow for Image Processing

Source: https://github.com/aiokitchen/aiomisc/blob/master/docs/source/worker_pool.md

Install the Pillow library, which is required for the image processing example. Use pip to install it.

```default
pip install Pillow
```

--------------------------------

### Run ASGI service with aiomisc

Source: https://github.com/aiokitchen/aiomisc/blob/master/docs/source/services/web.md

Start an ASGI application as a service using aiomisc. Requires aiohttp-asgi installed.

```python
import argparse

from fastapi import FastAPI

from aiomisc import entrypoint
from aiomisc.service.asgi import ASGIHTTPService, ASGIApplicationType

parser = argparse.ArgumentParser()
group = parser.add_argument_group('HTTP options')

group.add_argument("-l", "--address", default="::",
                   help="Listen HTTP address")
group.add_argument("-p", "--port", type=int, default=8080,
                   help="Listen HTTP port")


app = FastAPI()


@app.get("/")
async def root():
    return {"message": "Hello World"}


class REST(ASGIHTTPService):
    async def create_asgi_app(self) -> ASGIApplicationType:
        return app


arguments = parser.parse_args()
service = REST(address=arguments.address, port=arguments.port)

with entrypoint(service) as loop:
    loop.run_forever()
```

--------------------------------

### Install Dependencies

Source: https://github.com/aiokitchen/aiomisc/blob/master/examples/rpc/README.md

Install the necessary libraries for the RPC server and client implementation.

```bash
pip install aiomisc msgspec rich
```

--------------------------------

### Run aiohttp service with aiomisc

Source: https://github.com/aiokitchen/aiomisc/blob/master/docs/source/services/web.md

Start an aiohttp application as a service using aiomisc. Requires aiohttp installed.

```python
import aiohttp.web
import argparse
from aiomisc import entrypoint
from aiomisc.service.aiohttp import AIOHTTPService

parser = argparse.ArgumentParser()
group = parser.add_argument_group('HTTP options')

group.add_argument("-l", "--address", default="::",
                   help="Listen HTTP address")
group.add_argument("-p", "--port", type=int, default=8080,
                   help="Listen HTTP port")


async def handle(request):
    name = request.match_info.get('name', "Anonymous")
    text = "Hello, " + name
    return aiohttp.web.Response(text=text)


class REST(AIOHTTPService):
    async def create_application(self):
        app = aiohttp.web.Application()

        app.add_routes([
            aiohttp.web.get('/', handle),
            aiohttp.web.get('/{name}', handle)
        ])

        return app

arguments = parser.parse_args()
service = REST(address=arguments.address, port=arguments.port)

with entrypoint(service) as loop:
    loop.run_forever()
```

--------------------------------

### Install aiohttp-asgi for aiomisc

Source: https://github.com/aiokitchen/aiomisc/blob/master/docs/source/services/web.md

Install the aiohttp-asgi library or use aiomisc extras for ASGI integration.

```default
pip install aiohttp-asgi
```

```default
pip install aiomisc[asgi]
```

--------------------------------

### Start RPC Server

Source: https://github.com/aiokitchen/aiomisc/blob/master/examples/rpc_udp/README.md

Launch the RPC server using the Python module. The server will start listening on udp://[::]:15678.

```bash
$ python3 -m rpc.server
[12:26:34] INFO     Listening udp://[::]:15678                      utils.py:144
```

--------------------------------

### Install Dependencies

Source: https://github.com/aiokitchen/aiomisc/blob/master/examples/rpc_udp/README.md

Install the necessary libraries for the RPC server and client. This command installs aiomisc and msgspec.

```bash
pip install aiomisc msgspec
```

--------------------------------

### Define Plugin Entry Point in setup.py

Source: https://github.com/aiokitchen/aiomisc/blob/master/docs/source/plugins.md

Add 'aiomisc.plugins' entry to setup() in setup.py to make your plugin discoverable.

```python
# setup.py

setup(
    # ...
    entry_points={
        "aiomisc": ["myplugin = aiomisc_myplugin.plugin"]
    },
    # ...
)
```

--------------------------------

### Install DoH Server Manually

Source: https://github.com/aiokitchen/aiomisc/blob/master/examples/doh_server/README.md

This sequence of commands outlines the steps for installing the DoH server on a server manually. It involves creating a virtual environment, installing dependencies, copying the server script, and creating a symbolic link.

```shell
python3.8 -m venv /usr/share/python3/doh
/usr/share/python3/doh/bin/pip install -U pip \
    aiomisc~=15.6.1 \
    aiohttp~=3.8.0 \
    argclass~=0.6.0
cp doh_server.py /usr/share/python3/doh/bin/doh-server
ln -snf /usr/share/python3/doh/bin/doh-server /usr/local/bin/doh-server
```

--------------------------------

### Complete aiomisc Entrypoint Example with Configuration

Source: https://github.com/aiokitchen/aiomisc/blob/master/docs/source/entrypoint.md

Shows a comprehensive example of aiomisc.entrypoint with various configuration options such as pool size, logging level and format, signal catching, and shutdown timeout. Useful for fine-tuning application startup and shutdown behavior.

```python
import asyncio
import aiomisc
import logging
import signal

async def main():
    await asyncio.sleep(1)
    logging.info("Hello there")

with aiomisc.entrypoint(
    pool_size=2,
    log_level='info',
    log_format='color',                            # default when "rich" absent
    log_buffer_size=1024,                          # default
    log_flush_interval=0.2,                        # default
    log_config=True,                               # default
    debug=False,                                   # default
    catch_signals=(signal.SIGINT, signal.SIGTERM), # default
    shutdown_timeout=60,                           # default
) as loop:
    loop.run_until_complete(main())
```

--------------------------------

### Implementing and Running a Custom Service

Source: https://github.com/aiokitchen/aiomisc/blob/master/docs/source/services/index.md

Demonstrates how to create a custom service by inheriting from `aiomisc.Service`, implementing `start()`, and using `aiomisc.entrypoint` to run it. Ensure `self.start_event.set()` is called in `start()` if the task is intended to run indefinitely and notify the entrypoint. Avoid using `asyncio.get_event_loop()` directly; use `self.loop` instead.

```python
import asyncio
from threading import Event
from aiomisc import entrypoint, Service

event = Event()

class MyService(Service):
  async def start(self):
      # Send signal to entrypoint for continue running
      self.start_event.set()

      event.set()
      # Start service task
      await asyncio.sleep(3600)


with entrypoint(MyService()) as loop:
    assert event.is_set()
```

--------------------------------

### Run uvicorn service with aiomisc

Source: https://github.com/aiokitchen/aiomisc/blob/master/docs/source/services/web.md

Start an ASGI application via uvicorn as a service using aiomisc. Requires uvicorn installed.

```python
import argparse

from fastapi import FastAPI

from aiomisc import entrypoint
from aiomisc.service.uvicorn import UvicornApplication, UvicornService

parser = argparse.ArgumentParser()
group = parser.add_argument_group('HTTP options')

group.add_argument("-l", "--host", default="::",
                   help="Listen HTTP host")
group.add_argument("-p", "--port", type=int, default=8080,
                   help="Listen HTTP port")


app = FastAPI()


@app.get("/")
async def root():
    return {"message": "Hello World"}


class REST(UvicornService):
    async def create_application(self) -> UvicornApplication:
        return app


arguments = parser.parse_args()
service = REST(host=arguments.host, port=arguments.port)

with entrypoint(service) as loop:
    loop.run_forever()
```

--------------------------------

### Set Up Pre-commit Hooks

Source: https://github.com/aiokitchen/aiomisc/blob/master/docs/source/index.md

Execute this command to install and set up pre-commit hooks for the project. This helps maintain code quality and consistency.

```bash
poetry run pre-commit install
```

--------------------------------

### Install aiohttp for aiomisc

Source: https://github.com/aiokitchen/aiomisc/blob/master/docs/source/services/web.md

Install the aiohttp library or use aiomisc extras for aiohttp integration.

```default
pip install aiohttp
```

```default
pip install aiomisc[aiohttp]
```

--------------------------------

### Install aiomisc from GitHub

Source: https://github.com/aiokitchen/aiomisc/blob/master/docs/source/index.md

Install aiomisc directly from its GitHub repository using git or by downloading the master zip archive.

```bash
pip3 install git+https://github.com/aiokitchen/aiomisc.git
```

```bash
pip3 install \
    https://github.com/aiokitchen/aiomisc/archive/refs/heads/master.zip
```

--------------------------------

### Install Dependencies with Poetry

Source: https://github.com/aiokitchen/aiomisc/blob/master/docs/source/index.md

Use this command to install all project dependencies using Poetry. Ensure Poetry is installed and configured.

```bash
poetry install
```

--------------------------------

### Using entrypoint with Services

Source: https://github.com/aiokitchen/aiomisc/blob/master/docs/source/tutorial.md

Shows a basic usage of the `aiomisc.entrypoint` context manager to start and manage multiple services, ensuring proper termination upon interruption.

```python
import asyncio
import aiomisc

...

with aiomisc.entrypoint(
    OrdinaryService(),
    InfinityService()
) as loop:
    loop.run_forever()
```

--------------------------------

### Install aiomisc with uvloop extra

Source: https://github.com/aiokitchen/aiomisc/blob/master/docs/source/index.md

Install aiomisc with the 'uvloop' extra dependency for enhanced performance by using uvloop as the default event loop.

```bash
pip3 install "aiomisc[uvloop]"
```

--------------------------------

### Dynamically Starting and Stopping Services

Source: https://github.com/aiokitchen/aiomisc/blob/master/docs/source/entrypoint.md

Explains how to dynamically add, start, and stop services after the event loop has begun using aiomisc.entrypoint.get_current(). This is useful when service parameters are not known until runtime.

```python
import asyncio
import aiomisc
import logging

from aiomisc.service.periodic import PeriodicService

log = logging.getLogger(__name__)


class MyPeriodicService(PeriodicService):
    async def callback(self):
        log.info('Running periodic callback')


async def add_services():
    entrypoint = aiomisc.entrypoint.get_current()

    services = [
        MyPeriodicService(interval=2, delay=1),
        MyPeriodicService(interval=2, delay=0),
    ]

    await entrypoint.start_services(*services)
    await asyncio.sleep(10)
    await entrypoint.stop_services(*services)


with aiomisc.entrypoint() as loop:
    loop.create_task(add_services())
    loop.run_forever()
```

--------------------------------

### Start RPC Server

Source: https://github.com/aiokitchen/aiomisc/blob/master/examples/rpc/README.md

Run the RPC server using the command-line interface. The server will listen for incoming TCP connections.

```bash
$ python3 -m rpc.server
[22:25:24] INFO     Listening tcp://[::]:5678                       utils.py:144
[22:25:29] INFO     Start communication with tcp://[::1]:62121       spec.py:154
```

--------------------------------

### Start RPC Client

Source: https://github.com/aiokitchen/aiomisc/blob/master/examples/rpc_udp/README.md

Run the RPC client to connect to the server and execute requests. The client will report connection details and performance metrics.

```bash
$ python3 -m rpc.client
[12:26:54] INFO     Listening udp://[::]:63609                      utils.py:144
[12:26:55] INFO     Total executed 90000 requests on 1.103          client.py:34
           INFO     RPS: 81589.949                                  client.py:35
           INFO     Close connection                                client.py:37
```

--------------------------------

### Install aiomisc and aiomisc-pytest

Source: https://github.com/aiokitchen/aiomisc/blob/master/docs/source/pytest.md

Install both the aiomisc library and the aiomisc-pytest package for plugin registration.

```bash
pip install aiomisc aiomisc-pytest
```

--------------------------------

### Configuring Logging Before Entrypoint

Source: https://github.com/aiokitchen/aiomisc/blob/master/docs/source/entrypoint.md

Demonstrates configuring logging using `basic_config` before starting the `aiomisc.entrypoint`. It's safe to call `basic_config` multiple times.

```python
import logging

import aiomisc
from aiomisc_log import basic_config


basic_config(log_format="color")
logging.info("Hello from usual python")


async def main():
    logging.info("Hello from async python")


with aiomisc.entrypoint(log_format="color") as loop:
    loop.run_until_complete(main())
```

--------------------------------

### Start RPC Client

Source: https://github.com/aiokitchen/aiomisc/blob/master/examples/rpc/README.md

Execute the RPC client to establish a connection with the running server and initiate communication. This demonstrates client-side operations and performance metrics.

```bash
$ python3 -m rpc.client
[22:25:29] INFO     Connected to tcp://::1:5678                       tcp.py:106
           INFO     Start communication with tcp://[::1]:5678        spec.py:154
[22:25:40] INFO     Communication with tcp://[::1]:5678 finished     spec.py:176
[22:25:40] INFO     Total executed 1000000 requests on 10.623       client.py:38
           INFO     RPS: 94139.165                                  client.py:39
           INFO     Close connection                                client.py:40
```

--------------------------------

### Install uvicorn for aiomisc

Source: https://github.com/aiokitchen/aiomisc/blob/master/docs/source/services/web.md

Install the uvicorn library or use aiomisc extras for uvicorn integration.

```default
pip install uvicorn
```

```default
pip install aiomisc[uvicorn]
```

--------------------------------

### Install aiomisc from PyPI

Source: https://github.com/aiokitchen/aiomisc/blob/master/docs/source/index.md

Install the aiomisc library using pip. This is the standard method for obtaining the latest stable release.

```bash
pip3 install aiomisc
```

--------------------------------

### Install aiomisc with aiohttp extra

Source: https://github.com/aiokitchen/aiomisc/blob/master/docs/source/index.md

Install aiomisc with the 'aiohttp' extra dependency to enable support for running aiohttp applications.

```bash
pip3 install "aiomisc[aiohttp]"
```

--------------------------------

### Install aiomisc with DNS Support

Source: https://github.com/aiokitchen/aiomisc/blob/master/docs/source/services/periodic.md

Install the `aiomisc` library with the necessary dependencies for DNS server functionality using pip.

```bash
pip install aiomisc[dns]
```

--------------------------------

### Install aiomisc with uvloop support

Source: https://github.com/aiokitchen/aiomisc/blob/master/README.rst

Install the aiomisc library with uvloop support for enhanced performance. This is typically done using pip.

```bash
pip install aiomisc[uvloop]
```

--------------------------------

### Install Rich Library

Source: https://github.com/aiokitchen/aiomisc/blob/master/docs/source/logging.md

Install the Rich library, which is required for using the 'rich' or 'rich_tb' log formats.

```bash
pip install rich
```

--------------------------------

### Manual ThreadPoolExecutor Setup

Source: https://github.com/aiokitchen/aiomisc/blob/master/docs/source/threads.md

Manually set up a ThreadPoolExecutor for use with asyncio. This allows control over the number of threads in the pool. The `entrypoint` context manager also provides automatic setup.

```python
import asyncio
from aiomisc import ThreadPoolExecutor

loop = asyncio.get_event_loop()
thread_pool = ThreadPoolExecutor(4)
loop.set_default_executor(thread_pool)
```

--------------------------------

### Define Custom Services for Entrypoint

Source: https://github.com/aiokitchen/aiomisc/blob/master/docs/source/pytest.md

Override the `services` fixture to provide a list of `aiomisc.Service` instances to be started within the entrypoint.

```python
import aiomisc
import pytest


class MyService(aiomisc.Service):
    async def start(self) -> None:
        self.start_event.set()
        await asyncio.sleep(3600)


@pytest.fixture
def services():
    return [MyService()]
```

--------------------------------

### Install aiomisc with rich extra

Source: https://github.com/aiokitchen/aiomisc/blob/master/README.rst

Install aiomisc with the 'rich' extra dependencies to leverage the Rich library for enhanced logging output. This improves log readability.

```bash
pip install aiomisc[rich]
```

--------------------------------

### Install aiomisc with ASGI extra

Source: https://github.com/aiokitchen/aiomisc/blob/master/README.rst

Install aiomisc with the 'asgi' extra dependencies for running ASGI applications. This enables compatibility with ASGI servers and frameworks.

```bash
pip install aiomisc[asgi]
```

--------------------------------

### Install aiomisc with aiohttp extra

Source: https://github.com/aiokitchen/aiomisc/blob/master/README.rst

Install aiomisc with the 'aiohttp' extra dependencies for running aiohttp applications. This includes necessary packages for aiohttp integration.

```bash
pip install aiomisc[aiohttp]
```

--------------------------------

### Register POST_START Signal Callback

Source: https://github.com/aiokitchen/aiomisc/blob/master/docs/source/signal.md

Register an async callback for the POST_START signal, which occurs after all services have been started. Use this for sending startup notifications.

```python
from aiomisc import entrypoint, receiver

@receiver(entrypoint.POST_START)
async def startup_notifier(entrypoint, services):
  ...
```

```python
with entrypoint() as loop:
    loop.run_forever()
```

--------------------------------

### List Plugins with Human-Readable Output

Source: https://github.com/aiokitchen/aiomisc/blob/master/docs/source/plugins.md

Run 'python3 -m aiomisc.plugins -n' to get a human-readable list of plugins and their descriptions to stderr.

```default
$ python3 -m aiomisc.plugins -n
[12:25:57] INFO     Available 1 plugins.
           INFO     'systemd_watchdog' - Adds SystemD watchdog support to the entrypoint.
```

--------------------------------

### Async Fixture Example

Source: https://github.com/aiokitchen/aiomisc/blob/master/docs/source/pytest.md

Async fixtures work similarly to async test functions. Use `async def` for fixture definitions.

```python
import asyncio
import pytest


@pytest.fixture
async def my_fixture():
    await asyncio.sleep(0)
    yield "value"


async def test_with_fixture(my_fixture):
    assert my_fixture == "value"
```

--------------------------------

### Extend Service Initialization with Inheritance

Source: https://github.com/aiokitchen/aiomisc/blob/master/docs/source/tutorial.md

Demonstrates extending the Service initialization by overriding the __init__ method. This approach allows for custom initialization logic before the service starts.

```python
from typing import Any
import aiomisc

class HelloService(aiomisc.Service):
    def __init__(self, name: str = "world", **kwargs: Any):
        super().__init__(**kwargs)
        self.name = name

    async def start(self) -> Any:
        print(f"Hello {self.name}")

with aiomisc.entrypoint(
    HelloService(),
    HelloService(name="Guido")
) as loop:
    pass

# python hello.py
# <<< Hello world
# <<< Hello Guido
```

--------------------------------

### Implement SuicideService with RespawningProcessService

Source: https://github.com/aiokitchen/aiomisc/blob/master/docs/source/services/system.md

This example demonstrates how to create a service that will exit with a specific code after a delay. The RespawningProcessService will automatically restart it.

```python
import logging
from typing import Any

import aiomisc

from time import sleep


class SuicideService(aiomisc.service.RespawningProcessService):
    def in_process(self) -> Any:
        sleep(10)
        logging.warning("Goodbye mad world")
        exit(42)


if __name__ == '__main__':
    with aiomisc.entrypoint(SuicideService()) as loop:
        loop.run_forever()
```

--------------------------------

### Dynamically Manage Services with aiomisc

Source: https://context7.com/aiokitchen/aiomisc/llms.txt

Start and stop services dynamically after the entrypoint has begun. This is useful when service parameters are not known until runtime.

```python
import asyncio
import aiomisc
from aiomisc.service.periodic import PeriodicService
import logging

log = logging.getLogger(__name__)

class MonitorService(PeriodicService):
    def __init__(self, name: str, interval: float):
        super().__init__(interval=interval, delay=0)
        self.name = name

    async def callback(self):
        log.info(f'[{self.name}] Monitoring...')

async def manage_services():
    # Get the current entrypoint
    entrypoint = aiomisc.entrypoint.get_current()

    # Dynamically create and start services
    services = [
        MonitorService(name="cpu-monitor", interval=2),
        MonitorService(name="memory-monitor", interval=3),
    ]

    await entrypoint.start_services(*services)
    log.info("Dynamic services started")

    await asyncio.sleep(10)

    # Stop specific services
    await entrypoint.stop_services(*services)
    log.info("Dynamic services stopped")

with aiomisc.entrypoint(log_level="info", log_format="color") as loop:
    loop.create_task(manage_services())
    loop.run_until_complete(asyncio.sleep(15))
```

--------------------------------

### FastAPI Application Setup

Source: https://context7.com/aiokitchen/aiomisc/llms.txt

Defines a FastAPI application with Pydantic models for data validation and includes endpoints for root, getting/creating items, and listing all items.

```python
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from aiomisc import entrypoint
from aiomisc.service.asgi import ASGIHTTPService, ASGIApplicationType

# Create FastAPI application
app = FastAPI(title="My API")

class Item(BaseModel):
    name: str
    price: float
    quantity: int = 1

items_db = {}

@app.get("/")
async def root():
    return {"message": "Welcome to the API"}

@app.get("/items/{item_id}")
async def get_item(item_id: int):
    if item_id not in items_db:
        raise HTTPException(status_code=404, detail="Item not found")
    return items_db[item_id]

@app.post("/items/{item_id}")
async def create_item(item_id: int, item: Item):
    items_db[item_id] = item.dict()
    return {"created": True, "item": items_db[item_id]}

@app.get("/items")
async def list_items():
    return {"items": items_db}
```

--------------------------------

### Register PRE_START Signal Callback

Source: https://github.com/aiokitchen/aiomisc/blob/master/docs/source/signal.md

Register an async callback for the PRE_START signal, which occurs before any service has started. This is useful for tasks like preparing the database.

```python
from aiomisc import entrypoint, receiver

@receiver(entrypoint.PRE_START)
async def prepare_database(entrypoint, services):
  ...
```

```python
with entrypoint() as loop:
    loop.run_forever()
```

--------------------------------

### Async Fixture Example with aiomisc

Source: https://context7.com/aiokitchen/aiomisc/llms.txt

Illustrates the creation and usage of an asynchronous fixture that yields a resource and performs cleanup after the test. The fixture manages its state asynchronously.

```python
@pytest.fixture
async def async_resource():
    """Async fixture example"""
    resource = {"initialized": True}
    yield resource
    resource["initialized"] = False

async def test_with_async_fixture(async_resource):
    """Use async fixtures"""
    assert async_resource["initialized"] is True
```

--------------------------------

### Run aiomisc entrypoint with only logging configuration

Source: https://github.com/aiokitchen/aiomisc/blob/master/docs/source/tutorial.md

The entrypoint can be used solely for configuring logging and other parameters without starting any services. The event loop is managed within the 'with' block.

```python
import asyncio
import logging
import aiomisc


async def sleep_and_exit():
    logging.info("Started")
    await asyncio.sleep(1)
    logging.info("Exiting")


with aiomisc.entrypoint(log_level="info") as loop:
    loop.run_until_complete(sleep_and_exit())
```

--------------------------------

### Session-Scoped Database Pool Fixture

Source: https://github.com/aiokitchen/aiomisc/blob/master/docs/source/pytest.md

Example of a session-scoped fixture for a database pool, demonstrating how to leverage a session-scoped event loop for long-lived resources. Requires a compatible database driver and setup.

```python
# tests/integration/conftest.py (continued)
import pytest


@pytest.fixture(scope="session")
async def db_pool():
    pool = await create_pool(dsn="postgresql://localhost/test")
    yield pool
    await pool.close()
```

--------------------------------

### Entrypoint with Logging Configuration

Source: https://context7.com/aiokitchen/aiomisc/llms.txt

Shows how to use the `aiomisc.entrypoint` context manager to configure logging settings like level, format, buffer size, and flush interval for an entire application run.

```python
# Using entrypoint with logging configuration
async def app_main():
    logging.info("Application running")

with aiomisc.entrypoint(
    log_level='debug',
    log_format='color',
    log_buffer_size=2048,
    log_flush_interval=0.1
) as loop:
    loop.run_until_complete(app_main())
```

--------------------------------

### Run Coroutine with Cron Scheduling using aiomisc

Source: https://github.com/aiokitchen/aiomisc/blob/master/docs/source/utils.md

This example demonstrates how to schedule a coroutine function to run at a specified cron interval using aiomisc's CronCallback. Ensure croniter is installed.

```python
import asyncio
import time
from aiomisc import new_event_loop, CronCallback


async def cron_function():
    print("Hello")


if __name__ == '__main__':
    loop = new_event_loop()

    periodic = CronCallback(cron_function)

    # call it each second after that
    periodic.start(spec="* * * * * *")

    loop.run_forever()
```

--------------------------------

### Install croniter for aiomisc Cron Feature

Source: https://github.com/aiokitchen/aiomisc/blob/master/docs/source/utils.md

Install the croniter package separately or as an extra dependency when installing aiomisc to enable cron scheduling.

```shell
pip install croniter
```

```shell
pip install aiomisc[cron]
```

--------------------------------

### aiomisc.Service: Explicit Start and Stop Methods

Source: https://github.com/aiokitchen/aiomisc/blob/master/docs/source/tutorial.md

Define an `aiomisc.Service` with explicit `start` and `stop` methods for controlled execution. The service will run once, executing `start` and then `stop` upon completion or cancellation.

```python
import asyncio
import aiomisc
from typing import Any


class OrdinaryService(aiomisc.Service):
    async def start(self):
        # do some stuff
        ...

    async def stop(self, exception: Exception = None) -> Any:
        # do some stuff
        ...
```

--------------------------------

### aiomisc.Service: Infinite Start Coroutine

Source: https://github.com/aiokitchen/aiomisc/blob/master/docs/source/tutorial.md

Implement an `aiomisc.Service` with a `start` method that runs indefinitely. Use `self.start_event.set()` to signal readiness and `asyncio.sleep` for periodic tasks. The service stops when the `start` coroutine completes.

```python
import asyncio
import aiomisc


class InfinityService(aiomisc.Service):
    async def start(self):
        # Service is ready
        self.start_event.set()

        while True:
            # do some stuff
            await asyncio.sleep(1)
```

--------------------------------

### Full Entrypoint Configuration

Source: https://context7.com/aiokitchen/aiomisc/llms.txt

Configure thread pool size, logging, debug mode, signal handling, and shutdown timeout for comprehensive application lifecycle management.

```python
import asyncio
import logging
import signal
import aiomisc

log = logging.getLogger(__name__)

async def main():
    log.info('Application starting')
    await asyncio.sleep(3)
    log.info('Application finished')

# Full configuration with all options
with aiomisc.entrypoint(
    pool_size=4,                                    # Thread pool size
    log_level='info',                               # Logging level
    log_format='color',                             # Log format: color, json, rich, journald
    log_buffer_size=1024,                           # Log buffer size
    log_flush_interval=0.2,                         # Log flush interval in seconds
    log_config=True,                                # Enable logging configuration
    debug=False,                                    # Event loop debug mode
    catch_signals=(signal.SIGINT, signal.SIGTERM), # Signals to catch
    shutdown_timeout=60,                            # Graceful shutdown timeout
) as loop:
    loop.run_until_complete(main())
```

--------------------------------

### Simplified Async Execution with aiomisc.run()

Source: https://context7.com/aiokitchen/aiomisc/llms.txt

Use aiomisc.run() as a shortcut for asyncio.run() to automatically create and destroy an entrypoint, handling services and configuration.

```python
import asyncio
import aiomisc

async def main():
    loop = asyncio.get_event_loop()
    now = loop.time()
    await asyncio.sleep(0.1)
    print(f"Elapsed: {loop.time() - now:.2f}s")
    return "completed"

# Simple usage
result = aiomisc.run(main())
print(result)  # "completed"
```

```python
# With entrypoint configuration
aiomisc.run(main(), pool_size=8, log_level="debug", log_format="json")
```

--------------------------------

### List Available Plugins

Source: https://github.com/aiokitchen/aiomisc/blob/master/docs/source/plugins.md

Use 'python -m aiomisc.plugins' to list all discoverable plugins.

```default
$ python -m aiomisc.plugins
```

--------------------------------

### Basic Entrypoint Usage

Source: https://context7.com/aiokitchen/aiomisc/llms.txt

Use the entrypoint context manager for basic application lifecycle management, including event loop creation and graceful shutdown.

```python
import asyncio
import logging
import signal
import aiomisc

log = logging.getLogger(__name__)

async def main():
    log.info('Application starting')
    await asyncio.sleep(3)
    log.info('Application finished')

# Basic usage
with aiomisc.entrypoint() as loop:
    loop.run_until_complete(main())
```

--------------------------------

### Install aiomisc with carbon extra

Source: https://github.com/aiokitchen/aiomisc/blob/master/README.rst

Install aiomisc with the 'carbon' extra dependencies for sending metrics to Carbon, which is part of the Graphite monitoring system.

```bash
pip install aiomisc[carbon]
```

--------------------------------

### Define Plugin Entry Point in pyproject.toml

Source: https://github.com/aiokitchen/aiomisc/blob/master/docs/source/plugins.md

Configure plugin entry points in pyproject.toml for discoverability.

```ini
[tool.poetry.plugins.aiomisc]
myplugin = "aiomisc_myplugin.plugin"
```

--------------------------------

### Install aiomisc with cron extra

Source: https://github.com/aiokitchen/aiomisc/blob/master/README.rst

Install aiomisc with the 'cron' extra dependencies to use croniter for scheduling tasks. This enables cron-like scheduling capabilities.

```bash
pip install aiomisc[cron]
```

--------------------------------

### Service Initialization Notification with start_event

Source: https://github.com/aiokitchen/aiomisc/blob/master/docs/source/index.md

When a service's `start` method is the main payload, use `self.start_event.set()` to notify the `entrypoint` that initialization is complete. This allows the `entrypoint` to proceed after the service is ready.

```python
import asyncio
from threading import Event
from aiomisc import entrypoint, Service

event = Event()

class MyService(Service):
    async def start(self):
        # Send signal to entrypoint for continue running
        self.start_event.set()
        await asyncio.sleep(3600)


with entrypoint(MyService()) as loop:
    assert event.is_set()
```

--------------------------------

### Set Up Asynchronous DNS Server

Source: https://github.com/aiokitchen/aiomisc/blob/master/docs/source/services/periodic.md

Configure and run an asynchronous DNS server using `aiomisc` services. This involves defining DNS zones, records, and starting UDP and TCP server instances.

```python
from aiomisc import entrypoint
from aiomisc.service.dns import (
    DNSStore, DNSZone, TCPDNSServer, UDPDNSServer, records,
)

zone = DNSZone("test.")
zone.add_record(records.A.create("test.", "10.10.10.10"))
zone.add_record(records.AAAA.create("test.", "fd10::10"))

store = DNSStore()
store.add_zone(zone)
services = [
    UDPDNSServer(
        store=store, address="::1", port=5053,
    ),
    TCPDNSServer(
        store=store, address="::1", port=5053,
    ),
]

if __name__ == "__main__":
    with entrypoint(*services, log_level="debug") as loop:
        loop.run_forever()
```

--------------------------------

### GRPCService Initialization and Port Binding

Source: https://github.com/aiokitchen/aiomisc/blob/master/docs/source/services/grpc.md

Initializes a GRPCService, adds a custom GreeterServicer, and binds the service to multiple insecure and secure ports. Requires grpcio and aiomisc.

```python
import grpc

import aiomisc
from aiomisc.service.grpc_server import GRPCService


protos, services = grpc.protos_and_services("hello.proto")


class Greeter(services.GreeterServicer):
    async def SayHello(self, request, context):
        return protos.HelloReply(message='Hello, %s!' % request.name)


def main():
    grpc_service = GRPCService(compression=grpc.Compression.Gzip)
    services.add_GreeterServicer_to_server(
        Greeter(), grpc_service,
    )
    grpc_service.add_insecure_port('[::]:0')
    grpc_service.add_insecure_port('[::1]:0')
    grpc_service.add_insecure_port('127.0.0.1:0')
    grpc_service.add_insecure_port('localhost:0')
    grpc_service.add_secure_port('localhost:0', grpc.local_server_credentials())
    grpc_service.add_secure_port('[::]:0', grpc.local_server_credentials())

    with aiomisc.entrypoint(grpc_service) as loop:
        loop.run_forever()


if __name__ == '__main__':
    main()
```

--------------------------------

### Inherit from CronService and Register Callback

Source: https://github.com/aiokitchen/aiomisc/blob/master/docs/source/services/periodic.md

Subclasses CronService and registers a callback method within the start method. Ensures registration happens before the service starts.

```python
import aiomisc
from aiomisc.service.cron import CronService


class MyCronService(CronService):
    async def callback(self):
        log.info('Running cron callback')
        # ...

    async def start(self):
        self.register(self.callback, spec="0 * * * *")
        await super().start()

service = MyCronService()

with entrypoint(service) as loop:
    loop.run_forever()
```

--------------------------------

### Install aiomisc with raven extra

Source: https://github.com/aiokitchen/aiomisc/blob/master/README.rst

Install aiomisc with the 'raven' extra dependencies for sending exceptions to Sentry using the Raven client. This facilitates error tracking.

```bash
pip install aiomisc[raven]
```

--------------------------------

### Running Entrypoint from Async Code with Services

Source: https://github.com/aiokitchen/aiomisc/blob/master/docs/source/entrypoint.md

Illustrates how to use aiomisc.entrypoint within an already running event loop, typically when integrating with existing asyncio applications. It demonstrates starting a PeriodicService and waiting for the entrypoint to close.

```python
import asyncio
import aiomisc
import logging
from aiomisc.service.periodic import PeriodicService

log = logging.getLogger(__name__)

class MyPeriodicService(PeriodicService):
    async def callback(self):
        log.info('Running periodic callback')
        # ...

async def main():
    service = MyPeriodicService(interval=1, delay=0)  # once per minute

    # returns an entrypoint instance because event-loop
    # already running and might be get via asyncio.get_event_loop()
    async with aiomisc.entrypoint(service) as ep:
        try:
            await asyncio.wait_for(ep.closing(), timeout=1)
        except asyncio.TimeoutError:
            pass


asyncio.run(main())
```

--------------------------------

### List Plugins with Help Flag

Source: https://github.com/aiokitchen/aiomisc/blob/master/docs/source/plugins.md

View command-line options for listing plugins using the -h flag.

```default
$ python3 -m aiomisc.plugins -h
```

--------------------------------

### Install aiomisc with uvicorn extra

Source: https://github.com/aiokitchen/aiomisc/blob/master/README.rst

Install aiomisc with the 'uvicorn' extra dependencies for running ASGI applications using the Uvicorn server. This is useful for deploying ASGI apps.

```bash
pip install aiomisc[uvicorn]
```

--------------------------------

### Basic Event Loop with aiomisc Entrypoint

Source: https://github.com/aiokitchen/aiomisc/blob/master/docs/source/index.md

Use `aiomisc.entrypoint` to configure logging and run a simple asynchronous main function. This sets up the event loop and basic logging before executing your application's core logic.

```python
import asyncio
import logging

import aiomisc

log = logging.getLogger(__name__)

async def main():
    log.info('Starting')
    await asyncio.sleep(3)
    log.info('Exiting')


if __name__ == '__main__':
    with aiomisc.entrypoint(log_level="info", log_format="color") as loop:
        loop.run_until_complete(main())
```

--------------------------------

### Basic Database Insert and Select Tests

Source: https://github.com/aiokitchen/aiomisc/blob/master/docs/source/pytest.md

Illustrates simple asynchronous database operations for insertion and selection using a provided database pool fixture.

```python
async def test_insert(db_pool):
    async with db_pool.acquire() as conn:
        await conn.execute("INSERT INTO t VALUES (1)")


async def test_select(db_pool):
    async with db_pool.acquire() as conn:
        row = await conn.fetchrow("SELECT 1 AS n")
        assert row["n"] == 1
```

--------------------------------

### Defining a Custom aiomisc Service

Source: https://github.com/aiokitchen/aiomisc/blob/master/docs/source/index.md

Create a custom service by inheriting from `aiomisc.Service` and implementing `start` and `stop` methods. The `entrypoint` context manager handles starting and stopping these services gracefully.

```python
from aiomisc import entrypoint, Service

class MyService(Service):
    async def start(self):
        do_something_when_start()

    async def stop(self, exc):
        do_graceful_shutdown()


with entrypoint(MyService()) as loop:
    loop.run_forever()
```