from datetime import datetime
from email.utils import format_datetime
from pathlib import Path
+from urllib.parse import urlparse
import fastapi
from fastapi import Request, HTTPException, Query
database.cursor.execute("SELECT (SELECT COUNT(domain) FROM instances), (SELECT COUNT(domain) FROM instances WHERE software IN ('pleroma', 'mastodon', 'lemmy', 'friendica', 'misskey', 'peertube', 'takahe', 'gotosocial', 'brighteon', 'wildebeest', 'bookwyrm')), (SELECT COUNT(blocker) FROM blocks), (SELECT COUNT(domain) FROM instances WHERE last_error_details IS NOT NULL)")
row = database.cursor.fetchone()
- return {
+ return JSONResponse(status_code=200, content={
"known_instances" : row[0],
"supported_instances": row[1],
"blocks_recorded" : row[2],
"erroneous_instances": row[3],
- }
+ })
@router.get(config.get("base_url") + "/api/scoreboard.json", response_class=JSONResponse)
def api_scoreboard(mode: str, amount: int):
"score" : round(score)
})
- return scores
+ return JSONResponse(status_code=200, content=scores)
@router.get(config.get("base_url") + "/api/list.json", response_class=JSONResponse)
def api_list(request: Request, mode: str, value: str, amount: int):
domainlist = database.cursor.fetchall()
- return domainlist
+ return JSONResponse(status_code=200, content=dict(domainlist))
@router.get(config.get("base_url") + "/api/top.json", response_class=JSONResponse)
def api_index(request: Request, mode: str, value: str, amount: int):
if domain_data is None:
raise HTTPException(status_code=404, detail=f"domain='{domain}' not found")
- return domain_data
+ return JSONResponse(status_code=200, content=dict(domain_data))
@router.get(config.get("base_url") + "/api/mutual.json", response_class=JSONResponse)
def api_mutual(domains: list[str] = Query()):
# No known blocks
return JSONResponse(status_code=200, content={})
+@router.get(config.get("base_url") + "/.well-known/nodeinfo", response_class=JSONResponse)
+def wellknown_nodeinfo(request: Request):
+ components = urlparse(str(request.url))
+
+ return JSONResponse(status_code=200, content={
+ "links": ({
+ "rel" : "http://nodeinfo.diaspora.software/ns/schema/1.0",
+ "href": f"{components.scheme}://{config.get('hostname')}{config.get('base_url')}/nodeinfo/1.0"
+ })
+ })
+
+@router.get(config.get("base_url") + "/nodeinfo/1.0", response_class=JSONResponse)
+def nodeinfo_1_0(request: Request):
+ return JSONResponse(status_code=200, content={
+ "version": "1.0",
+ "software": {
+ "name": "FBA",
+ "version": "0.1",
+ },
+ "protocols": {
+ "inbound": (),
+ "outbound": (
+ "rss",
+ ),
+ },
+ "services": {
+ "inbound": (),
+ "outbound": (
+ "rss",
+ ),
+ },
+ "usage": {
+ "users": (),
+ },
+ "openRegistrations": False,
+ "metadata": {
+ "nodeName": "Fedi Block API",
+ "protocols": {
+ "inbound": (),
+ "outbound": (
+ "rss",
+ ),
+ },
+ "services": {
+ "inbound": (),
+ "outbound": (
+ "rss",
+ ),
+ },
+ "explicitContent": False,
+ },
+ })
+
+@router.get(config.get("base_url") + "/api/v1/instance/peers", response_class=JSONResponse)
+def api_peers(request: Request):
+ database.cursor.execute("SELECT domain FROM instances WHERE nodeinfo_url IS NOT NULL")
+
+ peers = list()
+ for row in database.cursor.fetchall():
+ peers.append(row["domain"])
+
+ return JSONResponse(status_code=200, content=peers)
+
@router.get(config.get("base_url") + "/scoreboard")
def scoreboard(request: Request, mode: str, amount: int):
response = None