Great — now that I see the full design, I can show you the clean, SQLAlchemy-style solution that:

✔️ Preserves Liskov Substitution Principle

✔️ Supports both sync and async callbacks (`fn_check_userpass`)

✔️ Does NOT use `nest_asyncio`

✔️ Does NOT break event loops

✔️ Allows `AuthServer` to remain fully synchronous

✔️ Adds async power only in the subclass

✔️ Uses greenlets (same technique SQLAlchemy uses)

And the best part:

You do NOT override authen_user() with async code. Instead, you add a new async method and use a greenlet bridge.

🚀 THE RIGHT ARCHITECTURE (SQLAlchemy style)

SQLAlchemy always does:

sync class (Engine)  → sync API only
async class (AsyncEngine) → async API

Async class WRAPS the sync class, it does not override methods with different sync/async signatures.

So your pattern should be:

AuthServer (sync)
AuthUserPassServer (sync)  <-- keep it sync!
AuthUserPassServerAsync (async wrapper)

Just like:

Engine
AsyncEngine

This is the only correct LSP-safe design.

💎 Final Result (Recommended SQLAlchemy-style API)

Users can do:

Sync usage

auth = AuthUserPassServer(check_fn)
token = auth.issue_token("alice", "pass")

Async usage

auth_async = AuthUserPassServerAsync(auth)
user = await auth_async.authen_user("alice", "pass")

This is exactly how SQLAlchemy does it.

🧩 Full Implementation

1️⃣ Base sync class (unchanged)

import jwt
from datetime import datetime, timedelta

class AuthServer:
    def __init__(self):
        self._access_token_expire_mins = 30
        self._secret_key = "supersecretkey123"
        self._algorithm = "HS256"

    def authen_user(self, *args, **kwargs) -> dict:
        raise NotImplementedError()

    def issue_token(self, *args, **kwargs):
        user_dict = self.authen_user(*args, **kwargs)
        if user_dict is None:
            raise Exception("User not authenticated")

        to_encode = user_dict.copy()
        expire = datetime.now() + timedelta(minutes=self._access_token_expire_mins)
        to_encode.update({"exp": expire})
        return jwt.encode(to_encode, self._secret_key, algorithm=self._algorithm)

    def verify_token(self, token: str) -> dict:
        return jwt.decode(token, self._secret_key, algorithms=[self._algorithm])

2️⃣ Sync subclass (`AuthUserPassServer`) — STAYS SYNC

import inspect

class AuthUserPassServer(AuthServer):
    def __init__(self, fn_check_userpass):
        super().__init__()
        assert fn_check_userpass is not None
        self.fn = fn_check_userpass

    def authen_user(self, username, password) -> dict:
        if inspect.iscoroutinefunction(self.fn):
            # Delegate async fn to the async wrapper class
            raise RuntimeError(
                "authen_user() received async function; "
                "use AuthUserPassServerAsync for async usage."
            )
        return self.fn(username, password)

This ensures sync API never deals with async directly → LSP preserved.

3️⃣ Async wrapper using greenlets (same as SQLAlchemy)

SQLAlchemy uses greenlets to run synchronous code inside async context.

We use the same trick.

import asyncio
from greenlet import greenlet

async def greenlet_spawn(func, *args, **kwargs):
    loop = asyncio.get_running_loop()
    fut = loop.create_future()

    def run():
        try:
            res = func(*args, **kwargs)
            loop.call_soon_threadsafe(fut.set_result, res)
        except BaseException as e:
            loop.call_soon_threadsafe(fut.set_exception, e)

    g = greenlet(run)
    g.switch()
    return await fut

4️⃣ Async wrapper class

class AuthUserPassServerAsync:
    def __init__(self, sync_server: AuthUserPassServer):
        self.sync_server = sync_server

    async def authen_user(self, username, password):
        fn = self.sync_server.fn

        # Case A: async function provided
        if inspect.iscoroutinefunction(fn):
            return await fn(username, password)

        # Case B: sync function provided → offload to greenlet
        return await greenlet_spawn(fn, username, password)

    async def issue_token(self, *args, **kwargs):
        return await greenlet_spawn(self.sync_server.issue_token, *args, **kwargs)

    async def verify_token(self, token):
        return await greenlet_spawn(self.sync_server.verify_token, token)

Now everything behaves exactly like:

Engine  <-->  AsyncEngine
Sync    <-->  Async

✔️ EXAMPLE USAGE

Sync usage

def check_user_sync(username, password):
    return {"sub": username} if password == "123" else None

server = AuthUserPassServer(check_user_sync)
print(server.issue_token("bob", "123"))

Async usage

async def check_user_async(username, password):
    await asyncio.sleep(0.1)
    return {"sub": username}

sync_server = AuthUserPassServer(check_user_async)
async_server = AuthUserPassServerAsync(sync_server)

user = await async_server.authen_user("alice", "pass")
print(user)