pb_api.py

# cron: 1 1 1 1 *
# const $ = new Env("pb数据库API");

import os  
from pocketbase import PocketBase  
from pocketbase.client import FileUpload  
from typing import BinaryIO, Optional, List, Dict  


class PbTalker:  
    def __init__(self) -> None:  
        # 1. base initialization  
        url = 'http://192.168.3.10:8090'  
        self.client = PocketBase(url)  
        auth = 'jufeng@example.com|dagesofu' 
        if not auth or "|" not in auth:  
            print("Warning: invalid email|password found, will handle with not auth, make sure you have set the collection rule by anyone")  
        else:  
            email, password = auth.split('|')  
            try:  
                admin_data = self.client.admins.auth_with_password(email, password)  
                if admin_data:  
                    print(f"Authenticated as admin - {email}")  
            except:  
                user_data = self.client.collection("users").auth_with_password(email, password)  
                if user_data:  
                    print(f"Authenticated as user - {email}")  
                else:  
                    raise Exception("pocketbase auth failed")  

    def read(self, collection_name: str, fields: Optional[List[str]] = None,  
                expand: Optional[List[str]] = None, filter: str = '', skiptotal: bool = True) -> list:  
        results = []  
        i = 1  
        while True:  
            try:  
                res = self.client.collection(collection_name).get_list(i, 500,  
                                                                       {"filter": filter,  
                                                                        "fields": ','.join(fields) if fields else '',  
                                                                        "expand": ','.join(expand) if expand else '',  
                                                                        "skiptotal": skiptotal})  

            except Exception as e:  
                print(f"pocketbase get list failed: {e}")  
                raise e  
            if not res.items:  
                break  
            for _res in res.items:  
                attributes = vars(_res)  
                results.append(attributes)  
            i += 1  
        return results  

    def add(self, collection_name: str, body: Dict) -> str:  
        try:  
            res = self.client.collection(collection_name).create(body)  
        except Exception as e:  
            print(f"pocketbase create failed: {e}")  
            return ''  
        return res.id  

    def update(self, collection_name: str, id: str, body: Dict) -> str:  
        try:  
            res = self.client.collection(collection_name).update(id, body)  
        except Exception as e:  
            print(f"pocketbase update failed: {e}")  
            return ''  
        return res.id  

    def delete(self, collection_name: str, id: str) -> bool:  
        try:  
            res = self.client.collection(collection_name).delete(id)  
        except Exception as e:  
            print(f"pocketbase delete failed: {e}")  
            return False  
        return bool(res)  

    def upload(self, collection_name: str, id: str, key: str, file_name: str, file: BinaryIO) -> str:  
        try:  
            res = self.client.collection(collection_name).update(id, {key: FileUpload((file_name, file))})  
        except Exception as e:  
            print(f"pocketbase upload failed: {e}")  
            return ''  
        return res.id  

    def view(self, collection_name: str, item_id: str, fields: Optional[List[str]] = None) -> Dict:  
        try:  
            res = self.client.collection(collection_name).get_one(item_id, {"fields": ','.join(fields) if fields else ''})  
            return vars(res)  
        except Exception as e:  
            print(f"pocketbase view item failed: {e}")  
            return {}