monoadmin/net-docs
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Initial commit

Browse Source
This commit is contained in:
monoadmin
2026-04-10 15:36:34 -07:00
commit 42d5299e93
11 changed files with 3918 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

6
backend/Dockerfile Normal file
View File

@@ -0,0 +1,6 @@
FROM python:3.12-slim
WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
COPY main.py .
CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8000"]

629
backend/main.py Normal file
View File

@@ -0,0 +1,629 @@
from fastapi import FastAPI, HTTPException
from fastapi.middleware.cors import CORSMiddleware
from sqlalchemy import create_engine, Column, String, Text, DateTime, ForeignKey
from sqlalchemy.orm import DeclarativeBase, Session, relationship
from pydantic import BaseModel, Field
from typing import Optional
import uuid, os, time
from datetime import datetime
DATABASE_URL = os.getenv("DATABASE_URL", "postgresql://netdoc:netdocpass@db/netdoc")
def get_engine():
for i in range(30):
try:
e = create_engine(DATABASE_URL)
with e.connect():
pass
return e
except Exception:
if i == 29:
raise
print(f"Waiting for DB... ({i+1}/30)")
time.sleep(2)
engine = get_engine()
class Base(DeclarativeBase):
pass
class Site(Base):
__tablename__ = "sites"
id = Column(String, primary_key=True)
name = Column(String, nullable=False)
location = Column(String)
contact = Column(String)
phone = Column(String)
email = Column(String)
notes = Column(Text)
created_at = Column(DateTime, default=datetime.utcnow)
devices = relationship("Device", back_populates="site", cascade="all, delete-orphan", order_by="Device.created_at")
subnets = relationship("Subnet", back_populates="site", cascade="all, delete-orphan", order_by="Subnet.created_at")
vlans = relationship("Vlan", back_populates="site", cascade="all, delete-orphan", order_by="Vlan.created_at")
isp_connections = relationship("IspConnection", back_populates="site", cascade="all, delete-orphan", order_by="IspConnection.created_at")
credentials = relationship("Credential", back_populates="site", cascade="all, delete-orphan", order_by="Credential.created_at")
notes_list = relationship("Note", back_populates="site", cascade="all, delete-orphan", order_by="Note.created_at")
class Device(Base):
__tablename__ = "devices"
id = Column(String, primary_key=True)
site_id = Column(String, ForeignKey("sites.id"), nullable=False)
hostname = Column(String, nullable=False)
ip = Column(String, nullable=False)
type = Column(String)
status = Column(String)
mac = Column(String)
vlan = Column(String)
model = Column(String)
location = Column(String)
notes = Column(Text)
created_at = Column(DateTime, default=datetime.utcnow)
site = relationship("Site", back_populates="devices")
class Subnet(Base):
__tablename__ = "subnets"
id = Column(String, primary_key=True)
site_id = Column(String, ForeignKey("sites.id"), nullable=False)
name = Column(String, nullable=False)
network = Column(String, nullable=False)
gateway = Column(String)
dhcp_start = Column(String)
dhcp_end = Column(String)
dns = Column(String)
vlan = Column(String)
notes = Column(Text)
created_at = Column(DateTime, default=datetime.utcnow)
site = relationship("Site", back_populates="subnets")
class Vlan(Base):
__tablename__ = "vlans"
id = Column(String, primary_key=True)
site_id = Column(String, ForeignKey("sites.id"), nullable=False)
vlan_num = Column(String)
name = Column(String, nullable=False)
purpose = Column(String)
ports = Column(String)
notes = Column(Text)
created_at = Column(DateTime, default=datetime.utcnow)
site = relationship("Site", back_populates="vlans")
class IspConnection(Base):
__tablename__ = "isp_connections"
id = Column(String, primary_key=True)
site_id = Column(String, ForeignKey("sites.id"), nullable=False)
label = Column(String, nullable=False)
provider = Column(String)
type = Column(String)
status = Column(String)
ips = Column(String)
gw = Column(String)
dl = Column(String)
ul = Column(String)
circuit = Column(String)
acct = Column(String)
phone = Column(String)
cost = Column(String)
renew = Column(String)
notes = Column(Text)
created_at = Column(DateTime, default=datetime.utcnow)
site = relationship("Site", back_populates="isp_connections")
class Credential(Base):
__tablename__ = "credentials"
id = Column(String, primary_key=True)
site_id = Column(String, ForeignKey("sites.id"), nullable=False)
label = Column(String, nullable=False)
cat = Column(String)
username = Column(String)
password = Column(String)
url = Column(String)
enable = Column(String)
notes = Column(Text)
created_at = Column(DateTime, default=datetime.utcnow)
site = relationship("Site", back_populates="credentials")
class Note(Base):
__tablename__ = "notes"
id = Column(String, primary_key=True)
site_id = Column(String, ForeignKey("sites.id"), nullable=False)
title = Column(String, nullable=False)
content = Column(Text, default="")
created_at = Column(DateTime, default=datetime.utcnow)
site = relationship("Site", back_populates="notes_list")
Base.metadata.create_all(engine)
app = FastAPI()
app.add_middleware(CORSMiddleware, allow_origins=["*"], allow_methods=["*"], allow_headers=["*"])
# ── Pydantic schemas ──────────────────────────────────────────────────────────
class SiteCreate(BaseModel):
name: str
location: Optional[str] = None
contact: Optional[str] = None
phone: Optional[str] = None
email: Optional[str] = None
notes: Optional[str] = None
class DeviceCreate(BaseModel):
hostname: str
ip: str
type: Optional[str] = None
status: Optional[str] = None
mac: Optional[str] = None
vlan: Optional[str] = None
model: Optional[str] = None
location: Optional[str] = None
notes: Optional[str] = None
class SubnetCreate(BaseModel):
name: str
network: str
gateway: Optional[str] = None
dhcpStart: Optional[str] = None
dhcpEnd: Optional[str] = None
dns: Optional[str] = None
vlan: Optional[str] = None
notes: Optional[str] = None
class VlanCreate(BaseModel):
id: str # VLAN number (1-4094)
name: str
purpose: Optional[str] = None
ports: Optional[str] = None
notes: Optional[str] = None
class IspCreate(BaseModel):
label: str
provider: Optional[str] = None
type: Optional[str] = None
status: Optional[str] = None
ips: Optional[str] = None
gw: Optional[str] = None
dl: Optional[str] = None
ul: Optional[str] = None
circuit: Optional[str] = None
acct: Optional[str] = None
phone: Optional[str] = None
cost: Optional[str] = None
renew: Optional[str] = None
notes: Optional[str] = None
class CredCreate(BaseModel):
label: str
cat: Optional[str] = None
user: Optional[str] = None
password: Optional[str] = None
url: Optional[str] = None
enable: Optional[str] = None
notes: Optional[str] = None
class NoteCreate(BaseModel):
title: str
content: Optional[str] = None
# ── Serializers ───────────────────────────────────────────────────────────────
def new_id():
return str(uuid.uuid4())
def s(val):
return val or ""
def site_to_dict(site: Site, full=False):
d = {
"_id": site.id,
"name": site.name,
"location": s(site.location),
"contact": s(site.contact),
"phone": s(site.phone),
"email": s(site.email),
"notes": s(site.notes),
}
if full:
d["devices"] = [device_to_dict(x) for x in site.devices]
d["subnets"] = [subnet_to_dict(x) for x in site.subnets]
d["vlans"] = [vlan_to_dict(x) for x in site.vlans]
d["isp"] = [isp_to_dict(x) for x in site.isp_connections]
d["creds"] = [cred_to_dict(x) for x in site.credentials]
d["notes"] = [note_to_dict(x) for x in site.notes_list]
return d
def device_to_dict(d: Device):
return {"_id": d.id, "hostname": d.hostname, "ip": d.ip, "type": s(d.type),
"status": s(d.status), "mac": s(d.mac), "vlan": s(d.vlan),
"model": s(d.model), "location": s(d.location), "notes": s(d.notes)}
def subnet_to_dict(x: Subnet):
return {"_id": x.id, "name": x.name, "network": x.network, "gateway": s(x.gateway),
"dhcpStart": s(x.dhcp_start), "dhcpEnd": s(x.dhcp_end),
"dns": s(x.dns), "vlan": s(x.vlan), "notes": s(x.notes)}
def vlan_to_dict(x: Vlan):
return {"_id": x.id, "id": s(x.vlan_num), "name": x.name,
"purpose": s(x.purpose), "ports": s(x.ports), "notes": s(x.notes)}
def isp_to_dict(x: IspConnection):
return {"_id": x.id, "label": x.label, "provider": s(x.provider), "type": s(x.type),
"status": s(x.status), "ips": s(x.ips), "gw": s(x.gw), "dl": s(x.dl),
"ul": s(x.ul), "circuit": s(x.circuit), "acct": s(x.acct),
"phone": s(x.phone), "cost": s(x.cost), "renew": s(x.renew), "notes": s(x.notes)}
def cred_to_dict(x: Credential):
return {"_id": x.id, "label": x.label, "cat": s(x.cat), "user": s(x.username),
"pass": s(x.password), "url": s(x.url), "enable": s(x.enable), "notes": s(x.notes)}
def note_to_dict(x: Note):
return {"_id": x.id, "title": x.title, "content": s(x.content)}
# ── Sites ─────────────────────────────────────────────────────────────────────
@app.get("/api/sites")
def list_sites():
with Session(engine) as sess:
return [site_to_dict(x) for x in sess.query(Site).order_by(Site.created_at).all()]
@app.post("/api/sites", status_code=201)
def create_site(data: SiteCreate):
with Session(engine) as sess:
site = Site(id=new_id(), **data.model_dump())
sess.add(site)
sess.commit()
sess.refresh(site)
return site_to_dict(site, full=True)
@app.get("/api/sites/{site_id}")
def get_site(site_id: str):
with Session(engine) as sess:
site = sess.query(Site).filter(Site.id == site_id).first()
if not site:
raise HTTPException(404, "Site not found")
return site_to_dict(site, full=True)
@app.delete("/api/sites/{site_id}")
def delete_site(site_id: str):
with Session(engine) as sess:
site = sess.query(Site).filter(Site.id == site_id).first()
if not site:
raise HTTPException(404)
sess.delete(site)
sess.commit()
return {"ok": True}
# ── Devices ───────────────────────────────────────────────────────────────────
@app.post("/api/sites/{site_id}/devices", status_code=201)
def create_device(site_id: str, data: DeviceCreate):
with Session(engine) as sess:
if not sess.query(Site).filter(Site.id == site_id).first():
raise HTTPException(404)
d = Device(id=new_id(), site_id=site_id, **data.model_dump())
sess.add(d)
sess.commit()
sess.refresh(d)
return device_to_dict(d)
@app.put("/api/sites/{site_id}/devices/{device_id}")
def update_device(site_id: str, device_id: str, data: DeviceCreate):
with Session(engine) as sess:
d = sess.query(Device).filter(Device.id == device_id, Device.site_id == site_id).first()
if not d:
raise HTTPException(404)
for k, v in data.model_dump().items():
setattr(d, k, v)
sess.commit()
sess.refresh(d)
return device_to_dict(d)
@app.delete("/api/sites/{site_id}/devices/{device_id}")
def delete_device(site_id: str, device_id: str):
with Session(engine) as sess:
d = sess.query(Device).filter(Device.id == device_id, Device.site_id == site_id).first()
if not d:
raise HTTPException(404)
sess.delete(d)
sess.commit()
return {"ok": True}
# ── Subnets ───────────────────────────────────────────────────────────────────
@app.post("/api/sites/{site_id}/subnets", status_code=201)
def create_subnet(site_id: str, data: SubnetCreate):
with Session(engine) as sess:
if not sess.query(Site).filter(Site.id == site_id).first():
raise HTTPException(404)
x = Subnet(id=new_id(), site_id=site_id, name=data.name, network=data.network,
gateway=data.gateway, dhcp_start=data.dhcpStart, dhcp_end=data.dhcpEnd,
dns=data.dns, vlan=data.vlan, notes=data.notes)
sess.add(x)
sess.commit()
sess.refresh(x)
return subnet_to_dict(x)
@app.put("/api/sites/{site_id}/subnets/{subnet_id}")
def update_subnet(site_id: str, subnet_id: str, data: SubnetCreate):
with Session(engine) as sess:
x = sess.query(Subnet).filter(Subnet.id == subnet_id, Subnet.site_id == site_id).first()
if not x:
raise HTTPException(404)
x.name = data.name
x.network = data.network
x.gateway = data.gateway
x.dhcp_start = data.dhcpStart
x.dhcp_end = data.dhcpEnd
x.dns = data.dns
x.vlan = data.vlan
x.notes = data.notes
sess.commit()
sess.refresh(x)
return subnet_to_dict(x)
@app.delete("/api/sites/{site_id}/subnets/{subnet_id}")
def delete_subnet(site_id: str, subnet_id: str):
with Session(engine) as sess:
x = sess.query(Subnet).filter(Subnet.id == subnet_id, Subnet.site_id == site_id).first()
if not x:
raise HTTPException(404)
sess.delete(x)
sess.commit()
return {"ok": True}
# ── VLANs ─────────────────────────────────────────────────────────────────────
@app.post("/api/sites/{site_id}/vlans", status_code=201)
def create_vlan(site_id: str, data: VlanCreate):
with Session(engine) as sess:
if not sess.query(Site).filter(Site.id == site_id).first():
raise HTTPException(404)
x = Vlan(id=new_id(), site_id=site_id, vlan_num=data.id,
name=data.name, purpose=data.purpose, ports=data.ports, notes=data.notes)
sess.add(x)
sess.commit()
sess.refresh(x)
return vlan_to_dict(x)
@app.put("/api/sites/{site_id}/vlans/{vlan_id}")
def update_vlan(site_id: str, vlan_id: str, data: VlanCreate):
with Session(engine) as sess:
x = sess.query(Vlan).filter(Vlan.id == vlan_id, Vlan.site_id == site_id).first()
if not x:
raise HTTPException(404)
x.vlan_num = data.id
x.name = data.name
x.purpose = data.purpose
x.ports = data.ports
x.notes = data.notes
sess.commit()
sess.refresh(x)
return vlan_to_dict(x)
@app.delete("/api/sites/{site_id}/vlans/{vlan_id}")
def delete_vlan(site_id: str, vlan_id: str):
with Session(engine) as sess:
x = sess.query(Vlan).filter(Vlan.id == vlan_id, Vlan.site_id == site_id).first()
if not x:
raise HTTPException(404)
sess.delete(x)
sess.commit()
return {"ok": True}
# ── ISP Connections ───────────────────────────────────────────────────────────
@app.post("/api/sites/{site_id}/isp", status_code=201)
def create_isp(site_id: str, data: IspCreate):
with Session(engine) as sess:
if not sess.query(Site).filter(Site.id == site_id).first():
raise HTTPException(404)
x = IspConnection(id=new_id(), site_id=site_id, **data.model_dump())
sess.add(x)
sess.commit()
sess.refresh(x)
return isp_to_dict(x)
@app.put("/api/sites/{site_id}/isp/{isp_id}")
def update_isp(site_id: str, isp_id: str, data: IspCreate):
with Session(engine) as sess:
x = sess.query(IspConnection).filter(IspConnection.id == isp_id, IspConnection.site_id == site_id).first()
if not x:
raise HTTPException(404)
for k, v in data.model_dump().items():
setattr(x, k, v)
sess.commit()
sess.refresh(x)
return isp_to_dict(x)
@app.delete("/api/sites/{site_id}/isp/{isp_id}")
def delete_isp(site_id: str, isp_id: str):
with Session(engine) as sess:
x = sess.query(IspConnection).filter(IspConnection.id == isp_id, IspConnection.site_id == site_id).first()
if not x:
raise HTTPException(404)
sess.delete(x)
sess.commit()
return {"ok": True}
# ── Credentials ───────────────────────────────────────────────────────────────
@app.post("/api/sites/{site_id}/creds", status_code=201)
def create_cred(site_id: str, data: CredCreate):
with Session(engine) as sess:
if not sess.query(Site).filter(Site.id == site_id).first():
raise HTTPException(404)
x = Credential(id=new_id(), site_id=site_id, label=data.label, cat=data.cat,
username=data.user, password=data.password, url=data.url,
enable=data.enable, notes=data.notes)
sess.add(x)
sess.commit()
sess.refresh(x)
return cred_to_dict(x)
@app.put("/api/sites/{site_id}/creds/{cred_id}")
def update_cred(site_id: str, cred_id: str, data: CredCreate):
with Session(engine) as sess:
x = sess.query(Credential).filter(Credential.id == cred_id, Credential.site_id == site_id).first()
if not x:
raise HTTPException(404)
x.label = data.label
x.cat = data.cat
x.username = data.user
x.password = data.password
x.url = data.url
x.enable = data.enable
x.notes = data.notes
sess.commit()
sess.refresh(x)
return cred_to_dict(x)
@app.delete("/api/sites/{site_id}/creds/{cred_id}")
def delete_cred(site_id: str, cred_id: str):
with Session(engine) as sess:
x = sess.query(Credential).filter(Credential.id == cred_id, Credential.site_id == site_id).first()
if not x:
raise HTTPException(404)
sess.delete(x)
sess.commit()
return {"ok": True}
# ── Notes ─────────────────────────────────────────────────────────────────────
@app.post("/api/sites/{site_id}/notes", status_code=201)
def create_note(site_id: str, data: NoteCreate):
with Session(engine) as sess:
if not sess.query(Site).filter(Site.id == site_id).first():
raise HTTPException(404)
x = Note(id=new_id(), site_id=site_id, title=data.title, content=data.content or "")
sess.add(x)
sess.commit()
sess.refresh(x)
return note_to_dict(x)
@app.put("/api/sites/{site_id}/notes/{note_id}")
def update_note(site_id: str, note_id: str, data: NoteCreate):
with Session(engine) as sess:
x = sess.query(Note).filter(Note.id == note_id, Note.site_id == site_id).first()
if not x:
raise HTTPException(404)
x.title = data.title
x.content = data.content or ""
sess.commit()
sess.refresh(x)
return note_to_dict(x)
@app.delete("/api/sites/{site_id}/notes/{note_id}")
def delete_note(site_id: str, note_id: str):
with Session(engine) as sess:
x = sess.query(Note).filter(Note.id == note_id, Note.site_id == site_id).first()
if not x:
raise HTTPException(404)
sess.delete(x)
sess.commit()
return {"ok": True}
# ── Export / Import ───────────────────────────────────────────────────────────
@app.get("/api/export")
def export_all():
with Session(engine) as sess:
sites = sess.query(Site).order_by(Site.created_at).all()
result = {"tenants": {}, "order": []}
for site in sites:
result["order"].append(site.id)
result["tenants"][site.id] = site_to_dict(site, full=True)
return result
@app.post("/api/import")
def import_all(data: dict):
with Session(engine) as sess:
for site_id in data.get("order", []):
t = data.get("tenants", {}).get(site_id, {})
if sess.query(Site).filter(Site.id == site_id).first():
continue
site = Site(id=site_id, name=t.get("name", "Imported Site"),
location=t.get("location"), contact=t.get("contact"),
phone=t.get("phone"), email=t.get("email"), notes=t.get("notes"))
sess.add(site)
for d in t.get("devices", []):
sess.add(Device(id=new_id(), site_id=site_id, hostname=d.get("hostname", ""),
ip=d.get("ip", ""), type=d.get("type"), status=d.get("status"),
mac=d.get("mac"), vlan=d.get("vlan"), model=d.get("model"),
location=d.get("location"), notes=d.get("notes")))
for x in t.get("subnets", []):
sess.add(Subnet(id=new_id(), site_id=site_id, name=x.get("name", ""),
network=x.get("network", ""), gateway=x.get("gateway"),
dhcp_start=x.get("dhcpStart"), dhcp_end=x.get("dhcpEnd"),
dns=x.get("dns"), vlan=x.get("vlan"), notes=x.get("notes")))
for x in t.get("vlans", []):
sess.add(Vlan(id=new_id(), site_id=site_id, vlan_num=x.get("id"),
name=x.get("name", ""), purpose=x.get("purpose"),
ports=x.get("ports"), notes=x.get("notes")))
for x in t.get("isp", []):
sess.add(IspConnection(id=new_id(), site_id=site_id, label=x.get("label", ""),
provider=x.get("provider"), type=x.get("type"),
status=x.get("status"), ips=x.get("ips"), gw=x.get("gw"),
dl=x.get("dl"), ul=x.get("ul"), circuit=x.get("circuit"),
acct=x.get("acct"), phone=x.get("phone"),
cost=x.get("cost"), renew=x.get("renew"), notes=x.get("notes")))
for x in t.get("creds", []):
sess.add(Credential(id=new_id(), site_id=site_id, label=x.get("label", ""),
cat=x.get("cat"), username=x.get("user"),
password=x.get("pass"), url=x.get("url"),
enable=x.get("enable"), notes=x.get("notes")))
for x in t.get("notes", []):
sess.add(Note(id=new_id(), site_id=site_id,
title=x.get("title", "Imported Note"),
content=x.get("content", "")))
sess.commit()
return {"ok": True}

5
backend/requirements.txt Normal file
View File

@@ -0,0 +1,5 @@
fastapi==0.111.0
uvicorn==0.29.0
sqlalchemy==2.0.30
psycopg2-binary==2.9.9
pydantic==2.7.1