Initial commit

.envrc

@@ -0,0 +1 @@
+use flake .

.gitignore

@@ -1,3 +1,5 @@
+.direnv
+
 # ---> Python
 # Byte-compiled / optimized / DLL files
 __pycache__/

flake.lock

@@ -0,0 +1,61 @@
+{
+  "nodes": {
+    "flake-parts": {
+      "inputs": {
+        "nixpkgs-lib": "nixpkgs-lib"
+      },
+      "locked": {
+        "lastModified": 1749398372,
+        "narHash": "sha256-tYBdgS56eXYaWVW3fsnPQ/nFlgWi/Z2Ymhyu21zVM98=",
+        "owner": "hercules-ci",
+        "repo": "flake-parts",
+        "rev": "9305fe4e5c2a6fcf5ba6a3ff155720fbe4076569",
+        "type": "github"
+      },
+      "original": {
+        "owner": "hercules-ci",
+        "repo": "flake-parts",
+        "type": "github"
+      }
+    },
+    "nixpkgs": {
+      "locked": {
+        "lastModified": 1750365781,
+        "narHash": "sha256-XE/lFNhz5lsriMm/yjXkvSZz5DfvKJLUjsS6pP8EC50=",
+        "owner": "nixos",
+        "repo": "nixpkgs",
+        "rev": "08f22084e6085d19bcfb4be30d1ca76ecb96fe54",
+        "type": "github"
+      },
+      "original": {
+        "owner": "nixos",
+        "ref": "nixos-unstable",
+        "repo": "nixpkgs",
+        "type": "github"
+      }
+    },
+    "nixpkgs-lib": {
+      "locked": {
+        "lastModified": 1748740939,
+        "narHash": "sha256-rQaysilft1aVMwF14xIdGS3sj1yHlI6oKQNBRTF40cc=",
+        "owner": "nix-community",
+        "repo": "nixpkgs.lib",
+        "rev": "656a64127e9d791a334452c6b6606d17539476e2",
+        "type": "github"
+      },
+      "original": {
+        "owner": "nix-community",
+        "repo": "nixpkgs.lib",
+        "type": "github"
+      }
+    },
+    "root": {
+      "inputs": {
+        "flake-parts": "flake-parts",
+        "nixpkgs": "nixpkgs"
+      }
+    }
+  },
+  "root": "root",
+  "version": 7
+}

flake.nix

@@ -0,0 +1,26 @@
+{
+  description = "A very basic flake";
+
+  inputs = {
+    nixpkgs.url = "github:nixos/nixpkgs?ref=nixos-unstable";
+    flake-parts.url = "github:hercules-ci/flake-parts";
+  };
+
+  outputs = inputs@{ flake-parts, ... }:
+    flake-parts.lib.mkFlake { inherit inputs; } (_: {
+      imports = [];
+      systems = [ "x86_64-linux" ];
+      perSystem = { pkgs, ... }: {
+        devShells.default = pkgs.mkShell {
+          buildInputs = with pkgs; [
+            python313
+            python313Packages.mypy
+            python313Packages.black
+            python313Packages.python-lsp-server
+            python313Packages.pylsp-mypy
+          ];
+        };
+      };
+    })
+  ;
+}

main.py

@@ -0,0 +1,3 @@
+#!/usr/bin/env python3
+
+from shackles import *

shackles/__init__.py

@@ -0,0 +1,144 @@
+import typing as t
+import abc
+import asyncio
+import enum
+
+P = t.ParamSpec("P")
+
+
+class Omit(enum.IntEnum):
+    OMIT = enum.auto()
+
+
+OMIT = Omit.OMIT
+
+
+class Endpoint[C, I, O](t.Protocol):
+    def __call__(self, ctx: C, args: I, /) -> t.Awaitable[O]: ...
+
+
+class Filter[C, I](Endpoint[C, I, bool], t.Protocol): ...
+
+
+# C I R N O
+class Handler[C, I, N, O](t.Protocol):
+    def __call__(self, ctx: C, args: I, next: N, /) -> t.Awaitable[O]: ...
+
+
+@t.runtime_checkable
+class Traversable(t.Protocol):
+    __leafs__: tuple[str, ...]
+
+
+def leafs_of(t: Traversable) -> tuple[str, ...]:
+    return t.__leafs__
+
+
+def traverse[A](what: Traversable, acc: A, f: t.Callable[[A, t.Any], A]) -> A:
+    """Traverse entire chain in depth."""
+    while True:
+        leafs = [(what, leafs_of(what))]
+        new_leafs = []
+        for what, leaf in leafs:
+            attr = getattr(what, leaf)  # type: ignore
+            acc = f(acc, attr)
+
+            if isinstance(attr, Traversable):
+                children = leafs_of(attr)
+                if children:
+                    new_leafs.append((attr, children))
+
+        if new_leafs:
+            leafs = new_leafs
+        else:
+            break
+
+    return acc
+
+
+def traverse_shallow[A](what: Traversable, acc: A, f: t.Callable[[A, t.Any], A]) -> A:
+    """Traverse only the first level of chain."""
+    for leaf in leafs_of(what):
+        attr = getattr(what, leaf)
+        acc = f(acc, attr)
+
+    return acc
+
+
+class Shard(Traversable, t.Protocol):
+    def apply[C, I, N, O](
+        self: Handler[C, I, N, O], rhs: N, /
+    ) -> "EComposable[C, I, O]":
+        """Apply continuation to the handler."""
+        return Apply[C, I, N, O](self, rhs)
+
+    def checked[C, I, O](
+        self: Endpoint[C, I, O],
+        filter: Filter[C, I],
+        err: Endpoint[C, I, O],
+    ) -> "EComposable[C, I, O]":
+        return EChecked[C, I, O](filter, self, err)
+
+    def then[C, I, N, O](
+        self, rhs: Handler[C, I, N, O], /
+    ) -> "Then[C, I, N, O, t.Self]":
+        """Connect two handlers, thus lhs will run after rhs."""
+        return Then[C, I, N, O, t.Self](self, rhs)
+
+
+class EComposable[C, I, O](Shard, Endpoint[C, I, O], t.Protocol):
+    """Composable endpoint."""
+
+
+class HComposable[C, I, N, O](Shard, Handler[C, I, N, O], t.Protocol):
+    """Composable handler."""
+
+
+class EChecked[C, I, O](Shard):
+    __slots__ = ("filter", "ok", "err")
+    __leafs__ = ("filter", "ok", "err")
+
+    def __init__(
+        self,
+        filter: Filter[C, I],
+        ok: Endpoint[C, I, O],
+        err: Endpoint[C, I, O],
+    ) -> None:
+        self.filter = filter
+        self.ok = ok
+        self.err = err
+
+    async def __call__(self, ctx: C, arg: I) -> O:
+        if await self.filter(ctx, arg):
+            return await self.ok(ctx, arg)
+        return await self.err(ctx, arg)
+
+
+class Then[C, I, N, O, L](Shard):
+    __slots__ = ("lhs", "rhs")
+    __leafs__ = ("lhs", "rhs")
+
+    def __init__(self, lhs: L, rhs: Handler[C, I, N, O]) -> None:
+        self.lhs = lhs
+        self.rhs = rhs
+
+    def __call__[Cl, In, On](
+        self: "Then[C, I, N, O, Handler[Cl, In, Endpoint[C, I, O], On]]",
+        ctx: Cl,
+        args: In,
+        next: N,
+    ) -> t.Awaitable[On]:
+        endpoint: Endpoint[C, I, O] = Apply(self.rhs, next)
+        return self.lhs(ctx, args, Apply(self.rhs, next))
+
+
+class Apply[C, I, N, O](Shard):
+    __slots__ = ("handler", "endpoint")
+    __leafs__ = ("handler", "endpoint")
+
+    def __init__(self, handler: Handler[C, I, N, O], endpoint: N) -> None:
+        self.handler = handler
+        self.endpoint = endpoint
+
+    def __call__(self, ctx: C, args: I) -> t.Awaitable[O]:
+        return self.handler(ctx, args, self.endpoint)

slonogram/_gen/__init__.py

@@ -0,0 +1 @@
+# Autogenerated thingies.

slonogram/bot.py

@@ -0,0 +1,7 @@
+import typing as t
+import dataclasses as dtc
+
+
+@dtc.dataclass
+class Bot:
+    token: str

slonogram/ctx.py

@@ -0,0 +1,7 @@
+import typing as t
+import dataclasses as dtc
+
+
+@dtc.dataclass
+class Ctx[R]:
+    request: R

slonogram/reqs/__init__.py

@@ -0,0 +1 @@
+

slonogram/reqs/raw.py

@@ -0,0 +1,22 @@
+import typing as t
+import dataclasses as dtc
+
+type Scalar = int | bool | float | str
+type Value = Scalar | dict[str, Scalar] | list[Scalar] | tuple[Scalar, ...]
+
+
+@dtc.dataclass
+class RawRequest:
+    method: str
+    data: dict[str, Value]
+    files: dict[str, t.BinaryIO] | None = None
+
+
+class Request(t.Protocol):
+    __method__: str
+
+    def into_raw(self) -> RawRequest: ...
+
+
+def method_of(r: Request) -> str:
+    return r.__method__

slonogram/stash.py

@@ -0,0 +1,44 @@
+import typing as t
+
+
+class Stash:
+    """A linked list of type:value maps. Generally you
+    want to put here some dependencies.
+    """
+
+    _ts: dict[t.Type, t.Any]
+    _next: "Stash | None"
+
+    def __init__(
+        self,
+        next: "Stash | None" = None,
+        _ts: dict[t.Type, t.Any] | None = None,
+    ) -> None:
+        if _ts is None:
+            self._ts = {}
+        else:
+            self._ts = _ts
+        self._next = next
+
+    def __getitem__[T](self, item: t.Type[T]) -> T:
+        try:
+            return self._ts[item]
+        except KeyError as e:
+            n = self._next
+            if n is None:
+                raise e from e
+            return n[item]
+
+    @classmethod
+    def empty(cls) -> t.Self:
+        return cls()
+
+    def mut_(self) -> "t.Self":
+        """Return instance which is safe to mutate."""
+        return type(self)(next=self)
+
+    def set[T](self, ty: t.Type[T], val: T) -> t.Self:
+        return type(self)(next=self._next, _ts={**self._ts, ty: val})
+
+    def append(self, rhs: "Stash") -> t.Self:
+        return type(self)(next=rhs, _ts=self._ts)